feat(hands,jobs): журнал прогонов дописывается, шторм новых видов глушится громко

beaver_agent/hands/komodo.py

@@ -37,6 +37,29 @@ Action = Literal[
 ANSI = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
 EXIT_MARK = "__KOMODO_EXIT_CODE:"
 NOT_RUNNING_LIMIT = 15
+UPDATES_LIMIT = 15
+UPDATES_MAX = 200
+
+LIST_LIMIT = 500
+LIST_PAGES = 20
+"""Komodo режет любой ``List*`` до своего ``default_pagination_limit`` (у нас
+50) и молчит об этом: ответ - просто короткий массив, неотличимый от полного.
+Поэтому листаем сами, а не верим первой странице."""
+
+RESOURCE_LISTS = (
+    "ListStacks",
+    "ListServers",
+    "ListRepos",
+    "ListResourceSyncs",
+    "ListProcedures",
+    "ListActions",
+    "ListAlerters",
+    "ListDeployments",
+    "ListBuilds",
+)
+"""Журнал операций называет ресурс id, а не именем: отсюда берём словарь."""
+
+SCOPE_NOTE = "список ограничен правами ключа: вне их Komodo ресурсы не показывает"
 
 
 @dataclass(frozen=True, slots=True)
@@ -65,6 +88,9 @@ class Komodo:
         tail: int = 100,
         update: str | None = None,
         server: str | None = None,
+        limit: int = UPDATES_LIMIT,
+        operation: str | None = None,
+        resource: str | None = None,
     ) -> str:
         """Прод через Komodo: посмотреть, а по просьбе Бобра - подействовать.
 
@@ -74,7 +100,11 @@ class Komodo:
         - containers: контейнеры сервера (server; без него - все серверы).
         - logs: логи стека (stack, service опционально, tail строк).
         - search_logs: поиск по логам стека (stack, terms - слова через пробел).
-        - updates: последние операции - кто, что, статус.
+        - updates: последние операции - кто, что, статус. limit записей
+          (по умолчанию 15, максимум 200), operation - подстрока типа операции
+          (deploy, sync, restart), resource - имя стека, сервера, репозитория.
+          Шапка говорит, какой отрезок журнала показан: журнал длинный, и
+          «под фильтр ничего не попало» - это не «операций больше не было».
         - update: подробности одной операции по id (update).
         - deploy / restart: стек целиком или один service; ждёт исхода до 3 мин.
         - exec: команда внутри контейнера (container, command; server, если имя
@@ -98,7 +128,7 @@ class Komodo:
                     _need(stack, "stack"), _need(terms, "terms"), service
                 )
             case "updates":
-                return await self._updates()
+                return await self._updates(limit, operation, resource)
             case "update":
                 return await self._update(_need(update, "update"))
             case "deploy":
@@ -118,7 +148,7 @@ class Komodo:
 
     async def _status(self) -> str:
         servers, stacks = await asyncio.gather(
-            self.read("ListServers", {}), self.read("ListStacks", {})
+            self.listing("ListServers"), self.listing("ListStacks")
         )
         lines = [f"{s['name']:24} {s['info']['state']}" for s in servers]
         bad = [s for s in stacks if s["info"]["state"] != "running"]
@@ -127,10 +157,15 @@ class Komodo:
             f"  {s['info']['state']:12} {s['name']} @{s['info'].get('server_name')}"
             for s in bad[:NOT_RUNNING_LIMIT]
         )
+        if len(bad) > NOT_RUNNING_LIMIT:
+            lines.append(
+                f"показаны {NOT_RUNNING_LIMIT} из {len(bad)}, остальные скрыты лимитом"
+            )
+        lines.append(SCOPE_NOTE)
         return "\n".join(lines)
 
     async def _stacks(self) -> str:
-        stacks = await self.read("ListStacks", {})
+        stacks = await self.listing("ListStacks")
         lines = []
         for s in sorted(stacks, key=lambda s: s["name"]):
             info = s["info"]
@@ -141,17 +176,18 @@ class Komodo:
                 f"{info['state']:12} {s['name']:28} @{info.get('server_name')}"
                 + ("  (есть незадеплоенные коммиты)" if behind else "")
             )
+        lines.append(f"стеков: {len(stacks)}; {SCOPE_NOTE}")
         return "\n".join(lines)
 
     async def _containers(self, server: str | None) -> str:
         names = (
             [server]
             if server
-            else [s["name"] for s in await self.read("ListServers", {})]
+            else [s["name"] for s in await self.listing("ListServers")]
         )
         lines = []
         for name in names:
-            containers = await self.read("ListDockerContainers", {"server": name})
+            containers = await self.listing("ListDockerContainers", {"server": name})
             lines.append(f"{name}:")
             lines.extend(
                 f"  {c['state']:10} {c['name']:36} {c.get('status', '')}"
@@ -177,9 +213,67 @@ class Komodo:
         )
         return self._clip(_log_text(log))
 
-    async def _updates(self) -> str:
-        updates = (await self.read("ListUpdates", {}))["updates"][:15]
-        return "\n".join(_update_line(u) for u in updates) or "операций нет"
+    async def _updates(
+        self, limit: int, operation: str | None, resource: str | None
+    ) -> str:
+        want = min(max(limit, 1), UPDATES_MAX)
+        names = await self._resource_names()
+        query: dict[str, Any] = {}
+        applied = []
+        if operation:
+            query["operation"] = {"$regex": re.escape(operation), "$options": "i"}
+            applied.append(f"операция ~{operation}")
+        if resource:
+            ids = [i for i, name in names.items() if resource.lower() in name.lower()]
+            if not ids:
+                return (
+                    f"ресурса ~{resource} ключу руки не видно: "
+                    "либо имя другое, либо он вне прав ключа"
+                )
+            query["target.id"] = {"$in": ids}
+            applied.append(f"ресурс ~{resource} ({len(ids)} шт)")
+        updates, more = await self._update_pages(query, want)
+
+        head = [f"операций: {len(updates)}"]
+        if updates:
+            head.append(
+                f"{_ts(updates[-1]['start_ts'])} … {_ts(updates[0]['start_ts'])}"
+            )
+        head.append("раньше в журнале есть ещё" if more else "это весь журнал")
+        lines = [", ".join(head)]
+        lines.append(
+            "фильтрует Komodo: " + "; ".join(applied)
+            if applied
+            else "фильтра нет: это просто хвост журнала"
+        )
+        if want != limit:
+            lines.append(f"запрошено {limit} записей, предел руки {want}")
+        unseen = sum(1 for u in updates if u["target"]["id"] not in names)
+        if unseen:
+            lines.append(
+                f"записей о невидимых ключу ресурсах: {unseen} - id вместо имени"
+            )
+        lines.extend(_update_line(u, names) for u in updates)
+        return "\n".join(lines)
+
+    async def _update_pages(
+        self, query: dict[str, Any], want: int
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """Страница ListUpdates - ровно 100 записей, limit там не работает."""
+        out: list[dict[str, Any]] = []
+        page: int | None = 0
+        while page is not None and len(out) < want:
+            body: dict[str, Any] = {"page": page}
+            if query:
+                body["query"] = query
+            answer = await self.read("ListUpdates", body)
+            out.extend(answer["updates"])
+            page = answer.get("next_page")
+        return out[:want], len(out) > want or page is not None
+
+    async def _resource_names(self) -> dict[str, str]:
+        lists = await asyncio.gather(*(self.listing(r) for r in RESOURCE_LISTS))
+        return {r["id"]: r["name"] for rows in lists for r in rows}
 
     async def _update(self, update_id: str) -> str:
         return self._clip(
@@ -224,12 +318,31 @@ class Komodo:
         return f"{header}\n{self._clip(output.strip())}"
 
     async def _server_of(self, container: str) -> str | None:
-        for s in await self.read("ListServers", {}):
-            containers = await self.read("ListDockerContainers", {"server": s["name"]})
+        for s in await self.listing("ListServers"):
+            containers = await self.listing(
+                "ListDockerContainers", {"server": s["name"]}
+            )
             if any(c["name"] == container for c in containers):
                 return s["name"]
         return None
 
+    async def listing(
+        self, request: str, params: dict[str, Any] | None = None
+    ) -> list[Any]:
+        """Любой ``List*`` целиком: страницами, пока Komodo не отдаст неполную.
+
+        Публичная: этим же листанием ходят джобы, второго такого не пишем.
+        """
+        out: list[Any] = []
+        for page in range(LIST_PAGES):
+            chunk = await self.read(
+                request, (params or {}) | {"page": page, "limit": LIST_LIMIT}
+            )
+            out.extend(chunk)
+            if len(chunk) < LIST_LIMIT:
+                break
+        return out
+
     async def _json(self, route: str, body: dict[str, Any]) -> Any:
         status, text = await self._post(route, body)
         if status != 200:
@@ -275,9 +388,10 @@ def _ts(ms: int) -> str:
     return datetime.fromtimestamp(ms / 1000, tz=UTC).strftime("%m-%d %H:%M")
 
 
-def _update_line(u: dict[str, Any]) -> str:
+def _update_line(u: dict[str, Any], names: dict[str, str]) -> str:
     ok = "ok" if u.get("success") else "FAIL"
-    head = f"{_ts(u['start_ts'])} {u['operation']:20} {u['status']:10} {ok}"
+    target = names.get(u["target"]["id"], u["target"]["id"])
+    head = f"{_ts(u['start_ts'])} {u['operation']:20} {target:26} {u['status']:10} {ok}"
     return f"{head} user={u.get('username')} id={u.get('id')}"
 
 

beaver_agent/jobs/__init__.py

@@ -10,6 +10,7 @@ from beaver_agent.jobs.closing import close_idle
 from beaver_agent.jobs.deploy import deploy
 from beaver_agent.jobs.komodo import komodo_alert
 from beaver_agent.jobs.leisure import tick
+from beaver_agent.jobs.logs import logs_watch
 from beaver_agent.jobs.memory import curate, memory
 from beaver_agent.jobs.rotation import rotate
 from beaver_agent.jobs.t3code import t3code_event
@@ -29,6 +30,11 @@ jobs = [
     Job("память", memory, cron="30 4 * * 0", critical=False),
     Job("голосовой", rotate_voice, cron="5 4 * * *", critical=False),
     Job("куратор", curate, cron="0 2,8,12,16,20 * * *", critical=False),
+    # Логи - раз в 3 часа и на ":40": на ":00" ротация и куратор. Чаще смысла
+    # мало (36 стеков = 36 чтений за проход), реже - хвост уезжает совсем.
+    # Проход 01:40 - первый после полуночи: он же несёт суточную сводку по
+    # длящемуся, чтобы фон не исчезал из виду, доложенный однажды.
+    Job("логи", logs_watch, cron="40 1,4,7,10,13,16,19,22 * * *", critical=False),
     Job("deploy", deploy, webhook=True),
     Job("komodo", komodo_alert, webhook=True, dedupe=False),
     Job("t3code", t3code_event, webhook=True, dedupe=False),

beaver_agent/jobs/logs.py

@@ -0,0 +1,146 @@
+"""Раз в 3 часа: хвосты логов своих стеков; в мастер уходит только новое.
+
+Мандат Бобра (20.09.2026): «можешь и логи периодами посматривать на предмет
+ошибок». Держит это механизм, а не память модели: код ходит по стекам своей
+области, а мастеру достаётся инжект только тогда, когда в хвостах появилось
+то, чего в прошлом проходе не было. Джоб ничего не чинит - только читает.
+
+Стеки берутся честным листанием (`Komodo.listing`), а не первой страницей
+`ListStacks`, и длина сверяется с `GetStacksSummary.total`: расхождение едет
+и в лог, и в инжект - неполный список хуже пустого.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+from beaver_agent.hands import KOMODO
+from beaver_agent.hands.komodo_logs import (
+    SCOPE,
+    TAIL,
+    Seen,
+    StackScan,
+    Sweep,
+    brief,
+    daily,
+    digest,
+    note,
+    read_stack,
+    record,
+)
+from beaver_agent.vault import STATE_DIR, ZONE
+
+if TYPE_CHECKING:
+    from beaver_gateway.jobs.scheduler import JobRun
+
+    from beaver_agent.hands.komodo import Komodo
+
+_log = logging.getLogger(__name__)
+
+STATE_FILE = STATE_DIR / "логи-флота.json"
+"""Состояние прохода: что уже видели и о чём уже сказали. Перезаписывается."""
+JOURNAL_FILE = STATE_DIR / "логи-флота.jsonl"
+"""Журнал прогонов, только дописывается: состояние - снимок, истории в нём нет,
+а пороги всплеска и глушения взяты из рассуждения и через неделю живой работы
+пересматриваются по числам. Читать: `jq` по строкам, поля - см. `record()`."""
+PARALLEL = 4
+"""Логи тянутся с periphery: четырёх параллельных хватает, больше - грубо."""
+
+
+async def _one(komodo: Komodo, stack: str, tail: int, gate: asyncio.Semaphore) -> Any:
+    async with gate:
+        try:
+            log = await komodo.read(
+                "GetStackLog",
+                {"stack": stack, "services": [], "tail": tail, "timestamps": True},
+            )
+        except Exception as exc:  # noqa: BLE001 - стек не прочитан, но проход идёт
+            _log.warning("логи: %s не прочитан: %s", stack, exc)
+            return StackScan(stack=stack, error=f"{type(exc).__name__}: {exc}"[:200])
+    return read_stack(stack, log)
+
+
+async def sweep(
+    komodo: Komodo,
+    *,
+    scope: tuple[str, ...] = SCOPE,
+    tail: int = TAIL,
+    now: datetime | None = None,
+) -> Sweep:
+    """Обойти стеки области и снять хвосты; Komodo молчит - так и скажем."""
+    at = now or datetime.now(UTC)
+    try:
+        stacks = await komodo.listing("ListStacks")
+    except Exception as exc:  # noqa: BLE001 - недоступный Komodo это не «чисто»
+        _log.exception("логи: Komodo не отдал список стеков")
+        return Sweep(
+            at=at,
+            unreachable=f"{type(exc).__name__}: {exc}"[:200],
+            tail=tail,
+            scope=scope,
+        )
+    summary: int | None = None
+    try:
+        summary = int((await komodo.read("GetStacksSummary", {}))["total"])
+    except Exception:  # noqa: BLE001 - сверка необязательна, проход без неё идёт
+        _log.warning("логи: GetStacksSummary не прочитан, сверить длину списка нечем")
+    if summary is not None and summary != len(stacks):
+        _log.error(
+            "логи: листание дало %s стеков, GetStacksSummary - %s", len(stacks), summary
+        )
+    mine = sorted(s["name"] for s in stacks if s["info"].get("server_name") in scope)
+    gate = asyncio.Semaphore(PARALLEL)
+    scans = await asyncio.gather(*(_one(komodo, name, tail, gate) for name in mine))
+    return Sweep(
+        at=at,
+        scans=tuple(scans),
+        in_scope=len(mine),
+        fleet=len(stacks),
+        summary=summary,
+        tail=tail,
+        scope=scope,
+    )
+
+
+async def logs_watch(run: JobRun) -> None:
+    """Проход по логам: состояние на диске, инжект - на новом и раз в сутки.
+
+    Инжекта два и они разные: обычный - про новое, всплески и отбой; суточный -
+    про длящееся, которое в проходах молчит. Первый проход суток у нас в 01:40,
+    он же и несёт сводку: в 00:00 ротация мастера, лезть туда незачем.
+    """
+    if KOMODO is None:
+        _log.info("логи: руки komodo нет, проход пропущен")
+        return
+    seen = Seen.load(STATE_FILE)
+    report = digest(seen, await sweep(KOMODO), ZONE)
+    if report.worth_telling and await run.inject_master(
+        brief(report, ZONE), urgency="normal", origin="логи"
+    ):
+        seen.mark_told(report)
+    if report.summary and await run.inject_master(
+        daily(report, ZONE), urgency="normal", origin="логи"
+    ):
+        seen.mark_daily(report)
+    seen.save()
+    row = record(report, ZONE) | {"pass_of_day": seen.passes_today}
+    try:
+        note(JOURNAL_FILE, row)
+    except OSError:
+        _log.exception("логи: строка прогона не дописалась в журнал")
+    _log.info(
+        "логи: новое %s, всплеск %s, стихло %s, длится %s (сводка %s), "
+        "не прочитано %s, стеков %s из %s, приглушено %s",
+        len(report.new),
+        len(report.surge),
+        len(report.quiet),
+        len(report.ongoing),
+        report.summary,
+        len(report.trouble),
+        len(report.sweep.read),
+        report.sweep.in_scope,
+        len(report.muted),
+    )

beaver_agent/memory/handouts.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import time
+from datetime import datetime
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from beaver_gateway.frontends.turn_record import slugify
@@ -15,12 +18,11 @@ from beaver_agent.vault import (
     PORTRAIT,
     STATE,
     VAULT,
+    ZONE,
     read,
 )
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from beaver_gateway.conversations.distill import DistillContext
     from beaver_gateway.conversations.rotation import HandoutContext
     from beaver_gateway.conversations.texts import NewDayContext, SeedContext
@@ -32,8 +34,67 @@ def latest_handout() -> Path | None:
     return files[-1] if files else None
 
 
+PROC_INIT = Path("/proc/1")
+"""mtime PID 1 - момент старта контейнера, проверено в бою."""
+PACKAGE = Path(__file__).resolve().parent.parent
+"""Корень пакета: свежесть кода меряем от себя, а не от пути в контейнере."""
+
+DEPLOY_MARK = "Выкат: процесс поднят {start} ({ago}), свежесть кода {code} - {verdict}."
+CODE_OLDER = "код старше старта, значит крутится то, что выкатывали"
+CODE_NEWER = "код НОВЕЕ старта: правка на диске есть, рестарта после неё не было"
+SAME_MINUTE = "выкат и правка в одну минуту"
+HOUR = 3600
+
+
+def _ago(seconds: float) -> str:
+    """Целыми часами, а меньше часа - минутами."""
+    if seconds < HOUR:
+        return f"{int(seconds // 60)} мин назад"
+    return f"{int(seconds // HOUR)} ч назад"
+
+
+def _clock(stamp: float) -> str:
+    return f"{datetime.fromtimestamp(stamp, ZONE):%Y-%m-%d %H:%M}"
+
+
+def deploy_mark(proc: Path = PROC_INIT, package: Path = PACKAGE) -> str | None:
+    """Когда подняли процесс и насколько свеж код, который в нём крутится.
+
+    Две цифры рядом: mtime PID 1 - старт контейнера, последний изменённый
+    `*.py` пакета - свежесть исходников. Порознь каждая врёт: код бывает
+    новее старта, и тогда правка на диске есть, а рестарта после неё не было.
+    Любая осечка (нет `/proc`, нет прав, пустой пакет) - None: сид не имеет
+    права упасть из-за этой отметки.
+    """
+    try:
+        started = proc.stat().st_mtime
+        sources = [
+            path.stat().st_mtime
+            for path in package.rglob("*.py")
+            if "__pycache__" not in path.parts
+        ]
+        if not sources:
+            return None
+        code = max(sources)
+        gap = code - started
+        if abs(gap) < 60:
+            verdict = SAME_MINUTE
+        elif gap > 0:
+            verdict = CODE_NEWER
+        else:
+            verdict = CODE_OLDER
+        return DEPLOY_MARK.format(
+            start=_clock(started),
+            ago=_ago(time.time() - started),
+            code=_clock(code),
+            verdict=verdict,
+        )
+    except (OSError, ValueError):
+        return None
+
+
 def seed_body(ctx: SeedContext) -> str | None:
-    """Утренний сид: последний хендаут, шапка состояния, портрет Бобра."""
+    """Утренний сид: хендаут, шапка состояния, портрет Бобра, отметка о выкате."""
     if ctx.seed != "morning":
         return None
     handout = latest_handout()
@@ -54,6 +115,9 @@ def seed_body(ctx: SeedContext) -> str | None:
             "Портрет Бобра (шапка "
             f"`{PORTRAIT.relative_to(VAULT)}`, лента - там же):\n{portrait}"
         )
+    mark = deploy_mark()
+    if mark:
+        parts.append(mark)
     return "\n\n".join(parts)
 
 

beaver_agent/vault.py

@@ -33,6 +33,10 @@ REPLIES = BEAVER / "реплики"
 CURATOR_JOURNAL = BEAVER / "куратор.md"
 LEISURE = BEAVER / "досуг.md"
 """Раскладка суток по тикам досуга: пишет её модель, механизм только читает."""
+STATE_DIR = Path(os.environ.get("BEAVER_STATE_DIR", BEAVER / ".состояние"))
+"""Машинное состояние джобов (не для чтения моделью): лежит в доме агента, а не
+в корне vault - структуру верхних папок ведёт владелец. Точка в начале: такой
+каталог Obsidian не показывает, а watcher смотрит только `.md`. Создаётся сам."""
 STATE = BEAVER / "состояние.md"
 STATE_MAX_LINES = 60
 ATTACHMENTS = BEAVER / "вложения"

tests/test_handouts.py

@@ -0,0 +1,72 @@
+import os
+from datetime import datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from beaver_agent.memory import handouts
+from beaver_agent.memory.handouts import deploy_mark
+
+ZONE = ZoneInfo("Europe/Warsaw")
+NOW = "2026-09-17 06:30"
+
+
+def _at(text: str) -> float:
+    return datetime.fromisoformat(text).replace(tzinfo=ZONE).timestamp()
+
+
+def _touch(path: Path, when: str) -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("", encoding="utf-8")
+    os.utime(path, (_at(when), _at(when)))
+    return path
+
+
+def _bundle(tmp_path: Path, started: str, code: str) -> tuple[Path, Path]:
+    proc = _touch(tmp_path / "proc" / "1", started)
+    package = tmp_path / "beaver_agent"
+    _touch(package / "memory" / "handouts.py", code)
+    _touch(package / "vault.py", "2026-09-10 12:00")
+    # мусор компилятора не должен выглядеть свежим кодом
+    _touch(package / "memory" / "__pycache__" / "handouts.py", "2026-09-17 06:00")
+    return proc, package
+
+
+@pytest.fixture(autouse=True)
+def _now(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(handouts.time, "time", lambda: _at(NOW))
+
+
+def test_code_older_than_start(tmp_path: Path) -> None:
+    proc, package = _bundle(tmp_path, "2026-09-16 03:11", "2026-09-16 03:03")
+    assert deploy_mark(proc, package) == (
+        "Выкат: процесс поднят 2026-09-16 03:11 (27 ч назад), свежесть кода "
+        "2026-09-16 03:03 - код старше старта, значит крутится то, что выкатывали."
+    )
+
+
+def test_code_newer_than_start(tmp_path: Path) -> None:
+    proc, package = _bundle(tmp_path, "2026-09-16 03:11", "2026-09-17 06:20")
+    assert deploy_mark(proc, package) == (
+        "Выкат: процесс поднят 2026-09-16 03:11 (27 ч назад), свежесть кода "
+        "2026-09-17 06:20 - код НОВЕЕ старта: правка на диске есть, "
+        "рестарта после неё не было."
+    )
+
+
+def test_same_minute(tmp_path: Path) -> None:
+    proc, package = _bundle(tmp_path, "2026-09-17 06:00", "2026-09-17 06:00:30")
+    assert deploy_mark(proc, package) == (
+        "Выкат: процесс поднят 2026-09-17 06:00 (30 мин назад), свежесть кода "
+        "2026-09-17 06:00 - выкат и правка в одну минуту."
+    )
+
+
+def test_no_source(tmp_path: Path) -> None:
+    proc, package = _bundle(tmp_path, "2026-09-16 03:11", "2026-09-16 03:03")
+    assert deploy_mark(tmp_path / "нет", package) is None
+    assert deploy_mark(proc, tmp_path / "пусто") is None
+    only_cache = tmp_path / "кеш"
+    _touch(only_cache / "__pycache__" / "handouts.py", "2026-09-17 06:00")
+    assert deploy_mark(proc, only_cache) is None

tests/test_komodo.py

@@ -49,3 +49,138 @@ def test_clip_keeps_head_and_tail():
     k = Komodo(url="u", key="k", secret="s", max_chars=30)
     out = k._clip("a" * 100)
     assert out.startswith("a" * 20) and out.endswith("a" * 10) and "обрезано" in out
+
+
+def fleet(answers, calls=None):
+    """Komodo, отвечающий заготовками; calls копит запросы к нему."""
+
+    class Fake(Komodo):
+        async def read(self, request, params):
+            if calls is not None:
+                calls.append((request, params))
+            answer = answers.get(request, [])
+            return answer(params) if callable(answer) else answer
+
+    return Fake(url="u", key="k", secret="s")
+
+
+def a_stack(name, state="running", stack_id=None):
+    return {
+        "id": stack_id or name,
+        "name": name,
+        "info": {"state": state, "server_name": "dell-home"},
+    }
+
+
+def an_update(operation="DeployStack", target_id="beaver-agent", ts=1_700_000_000_000):
+    return {
+        "id": f"upd-{ts}",
+        "operation": operation,
+        "start_ts": ts,
+        "success": True,
+        "status": "Complete",
+        "username": "hh",
+        "target": {"type": "Stack", "id": target_id},
+    }
+
+
+async def test_a_short_page_is_not_the_whole_list():
+    # Komodo режет List* до default_pagination_limit и молчит об этом.
+    pages = [
+        [a_stack(f"s{i}") for i in range(500)],
+        [a_stack("zv-import"), a_stack("veris-runner")],
+    ]
+    calls = []
+    komodo = fleet({"ListStacks": lambda p: pages[p["page"]]}, calls)
+    out = await komodo._stacks()
+    assert [p["page"] for r, p in calls if r == "ListStacks"] == [0, 1]
+    assert "zv-import" in out and "стеков: 502" in out
+
+
+async def test_status_says_when_the_not_running_list_is_cut():
+    bad = [a_stack(f"b{i}", state="unknown") for i in range(23)]
+    komodo = fleet({"ListStacks": bad, "ListServers": []})
+    out = await komodo._status()
+    assert "стеков: 23, не running: 23" in out
+    assert "показаны 15 из 23, остальные скрыты лимитом" in out
+
+    komodo = fleet({"ListStacks": bad[:3], "ListServers": []})
+    assert "скрыты лимитом" not in await komodo._status()
+
+
+async def test_both_listings_say_that_they_are_limited_by_rights():
+    komodo = fleet({"ListStacks": [a_stack("beaver-agent")], "ListServers": []})
+    assert "ограничен правами ключа" in await komodo._status()
+    assert "ограничен правами ключа" in await komodo._stacks()
+
+
+async def test_updates_carry_the_shown_span_and_say_there_is_more():
+    page = {
+        "updates": [an_update(ts=1_700_000_000_000 - i * 60_000) for i in range(100)]
+    }
+    seen = {"ListStacks": [a_stack("beaver-agent")]}
+    komodo = fleet(seen | {"ListUpdates": page | {"next_page": 1}})
+    head, second, *lines = (await komodo._updates(15, None, None)).splitlines()
+    assert head == "операций: 15, 11-14 21:59 … 11-14 22:13, раньше в журнале есть ещё"
+    assert second == "фильтра нет: это просто хвост журнала"
+    assert len(lines) == 15
+
+    komodo = fleet(
+        seen | {"ListUpdates": {"updates": page["updates"][:3], "next_page": None}}
+    )
+    assert (await komodo._updates(15, None, None)).startswith(
+        "операций: 3, 11-14 22:11 … 11-14 22:13, это весь журнал"
+    )
+
+
+async def test_the_filter_goes_to_komodo_as_a_query():
+    calls = []
+    komodo = fleet(
+        {
+            "ListStacks": [a_stack("beaver-agent", stack_id="id-1")],
+            "ListUpdates": {
+                "updates": [an_update(target_id="id-1")],
+                "next_page": None,
+            },
+        },
+        calls,
+    )
+    out = await komodo._updates(15, "deploy", "beaver")
+    (params,) = [p for r, p in calls if r == "ListUpdates"]
+    assert params["query"] == {
+        "operation": {"$regex": "deploy", "$options": "i"},
+        "target.id": {"$in": ["id-1"]},
+    }
+    assert "фильтрует Komodo: операция ~deploy; ресурс ~beaver (1 шт)" in out
+    # имя ресурса в строке, а не id
+    assert "beaver-agent" in out and "id-1 " not in out
+
+
+async def test_an_unknown_name_is_said_out_loud_instead_of_filtering_by_nothing():
+    komodo = fleet({"ListStacks": [a_stack("beaver-agent")]})
+    out = await komodo._updates(15, None, "obsidian")
+    assert out == (
+        "ресурса ~obsidian ключу руки не видно: либо имя другое, либо он вне прав ключа"
+    )
+
+
+async def test_a_target_the_key_cannot_see_is_counted_and_shown_by_id():
+    komodo = fleet(
+        {
+            "ListStacks": [a_stack("beaver-agent", stack_id="id-1")],
+            "ListUpdates": {
+                "updates": [an_update(target_id="id-1"), an_update(target_id="gone")],
+                "next_page": None,
+            },
+        }
+    )
+    out = await komodo._updates(15, None, None)
+    assert "записей о невидимых ключу ресурсах: 1 - id вместо имени" in out
+    assert "gone" in out
+
+
+async def test_the_depth_is_clamped_out_loud():
+    komodo = fleet({"ListUpdates": {"updates": [], "next_page": None}})
+    assert "запрошено 5000 записей, предел руки 200" in await komodo._updates(
+        5000, None, None
+    )

tests/test_komodo_logs.py

@@ -0,0 +1,813 @@
+import json
+from datetime import UTC, datetime, timedelta
+
+from beaver_agent.hands.komodo import Komodo
+from beaver_agent.hands.komodo_logs import (
+    KEEP,
+    MUTE_KINDS,
+    MUTE_SHOW,
+    SELF_MARK,
+    SURGE_CEILING,
+    SURGE_FACTOR,
+    SURGE_MIN,
+    Seen,
+    StackScan,
+    Sweep,
+    brief,
+    daily,
+    digest,
+    grouped,
+    note,
+    read_stack,
+    record,
+)
+from beaver_agent.jobs.logs import sweep as sweep_fleet
+
+NOW = datetime(2026, 9, 20, 18, 0, tzinfo=UTC)
+
+
+def log(*lines):
+    return {"stdout": "\n".join(lines), "stderr": ""}
+
+
+def line(service, at, text):
+    return f"{service}  | {at}Z {text}"
+
+
+def a_stack(name, server="dell-home"):
+    return {
+        "id": name,
+        "name": name,
+        "info": {"state": "running", "server_name": server},
+    }
+
+
+def fleet(answers, calls=None):
+    """Komodo, отвечающий заготовками; calls копит запросы к нему."""
+
+    class Fake(Komodo):
+        async def read(self, request, params):
+            if calls is not None:
+                calls.append((request, params))
+            answer = answers.get(request, [])
+            return answer(params) if callable(answer) else answer
+
+    return Fake(url="u", key="k", secret="s")
+
+
+def swept(scans, **kw):
+    kw.setdefault("in_scope", len(scans))
+    kw.setdefault("fleet", len(scans))
+    return Sweep(at=kw.pop("at", NOW), scans=tuple(scans), **kw)
+
+
+KNOWN = ("beaver-agent", "gitea", "democalc", "mix", "healthbot")
+
+
+def seen(tmp_path, *, fresh=False, known=KNOWN):
+    """Состояние с уже снятой базовой линией: проход не первый."""
+    state = Seen.load(tmp_path / "логи.json")
+    if not fresh:
+        state.fresh = False
+        state.baselined = set(known)
+    return state
+
+
+def test_a_line_gives_its_service_its_time_and_its_signal():
+    scan = read_stack(
+        "beaver-agent",
+        log(
+            line("beaver-gateway", "2026-09-20T17:40:01.123456789", "ERROR boom"),
+            line("beaver-gateway", "2026-09-20T17:41:02.000000000", "ERROR boom"),
+            line("beaver-obsidian", "2026-09-20T17:42:03.000000000", "Fully synced"),
+        ),
+    )
+    (hit,) = scan.hits
+    assert hit.service == "beaver-gateway"
+    assert hit.signal == "ошибка"
+    assert hit.count == 2
+    assert hit.at == datetime(2026, 9, 20, 17, 41, 2, tzinfo=UTC)
+    assert scan.first == datetime(2026, 9, 20, 17, 40, 1, 123456, tzinfo=UTC)
+    assert scan.last == datetime(2026, 9, 20, 17, 42, 3, tzinfo=UTC)
+
+
+def test_the_same_trouble_with_other_numbers_is_the_same_trouble():
+    scan = read_stack(
+        "gitea",
+        log(
+            line("gitea", "2026-09-20T17:40:01.0", "ERROR request 8871 failed"),
+            line("gitea", "2026-09-20T17:40:09.0", "ERROR request 9134 failed"),
+        ),
+    )
+    (hit,) = scan.hits
+    assert hit.count == 2
+
+
+def test_every_named_signal_is_caught():
+    texts = (
+        "CRITICAL disk is gone",
+        "panic: runtime error",
+        "container OOMKilled",
+        "Traceback (most recent call last):",
+        "service exited with code 137",
+        "beaver-gateway Restarting (1) 3 seconds ago",
+        "container is unhealthy",
+        "ERROR plain",
+    )
+    scan = read_stack(
+        "mix",
+        log(*(line("svc", f"2026-09-20T17:4{i}:00.0", t) for i, t in enumerate(texts))),
+    )
+    assert {h.signal for h in scan.hits} == {
+        "критично",
+        "паника",
+        "oom",
+        "traceback",
+        "падение",
+        "рестарт-цикл",
+        "healthcheck",
+        "ошибка",
+    }
+    assert "exited with code 0" not in {h.sample for h in scan.hits}
+
+
+def test_a_clean_exit_and_our_own_inject_are_not_trouble():
+    scan = read_stack(
+        "beaver-agent",
+        log(
+            line("svc", "2026-09-20T17:40:00.0", "worker exited with code 0"),
+            line("beaver-gateway", "2026-09-20T17:41:00.0", f"{SELF_MARK}: ERROR boom"),
+        ),
+    )
+    assert scan.hits == ()
+
+
+def test_komodos_own_error_is_not_a_clean_stack():
+    scan = read_stack("gitea", {"error": "periphery unreachable"})
+    assert not scan.ok
+    assert scan.error == "periphery unreachable"
+    assert Sweep(at=NOW, scans=(scan,), in_scope=1, fleet=1).trouble == (
+        "gitea: periphery unreachable",
+    )
+
+
+def test_the_first_pass_only_takes_the_baseline(tmp_path):
+    state = seen(tmp_path, fresh=True)
+    scan = read_stack(
+        "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR старая беда"))
+    )
+    report = digest(state, swept([scan]))
+    assert report.cold and report.baseline == 1
+    assert report.new == ()
+    text = brief(report)
+    assert "базовую линию" in text
+    assert "прочитано 1 стеков из 1" in text
+
+
+def test_only_the_new_line_wakes_the_master(tmp_path):
+    old = line("svc", "2026-09-20T17:40:00.0", "ERROR молотит третьи сутки")
+    state = seen(tmp_path, fresh=True)
+    first = digest(state, swept([read_stack("beaver-agent", log(old))]))
+    state.mark_told(first)
+    state.save()
+
+    fresh_line = line("svc", "2026-09-20T18:50:00.0", "CRITICAL диск кончился")
+    state = Seen.load(tmp_path / "логи.json")
+    report = digest(
+        state,
+        swept(
+            [read_stack("beaver-agent", log(old, old, fresh_line))],
+            at=NOW + timedelta(hours=3),
+        ),
+    )
+    (new,) = report.new
+    assert new.signal == "критично"
+    text = brief(report)
+    assert "диск кончился" in text
+    assert "молотит третьи сутки" not in text
+
+
+def test_a_repeat_carries_how_many_times(tmp_path):
+    noisy = [
+        line("svc", f"2026-09-20T17:4{i}:00.0", "ERROR раз за разом") for i in range(5)
+    ]
+    report = digest(seen(tmp_path), swept([read_stack("beaver-agent", log(*noisy))]))
+    assert "×5" in brief(report)
+
+
+def test_a_trouble_that_stopped_is_dated_by_the_event_not_by_the_look(tmp_path):
+    was = line("svc", "2026-09-20T12:03:00.0", "ERROR было и прошло")
+    state = seen(tmp_path)
+    state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))])))
+
+    later = NOW + timedelta(hours=6)
+    report = digest(state, swept([read_stack("beaver-agent", log())], at=later))
+    (quiet,) = report.quiet
+    assert quiet.last == datetime(2026, 9, 20, 12, 3, tzinfo=UTC)
+    text = brief(report)
+    assert "последнее в 20.09 12:03" in text
+    assert f"{later:%H:%M}" not in text.split("\n", 1)[1]
+
+
+def test_a_stack_that_did_not_answer_is_not_a_stack_that_went_quiet(tmp_path):
+    was = line("svc", "2026-09-20T12:03:00.0", "ERROR было")
+    state = seen(tmp_path)
+    state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))])))
+
+    report = digest(
+        state,
+        swept(
+            [StackScan(stack="beaver-agent", error="timeout")],
+            at=NOW + timedelta(hours=3),
+        ),
+    )
+    assert report.quiet == ()
+    assert report.trouble == ("beaver-agent: timeout",)
+    assert "beaver-agent: timeout" in brief(report)
+
+
+def test_an_undelivered_inject_leaves_the_trouble_new(tmp_path):
+    scan = read_stack(
+        "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
+    )
+    state = seen(tmp_path)
+    first = digest(state, swept([scan]))
+    assert len(first.new) == 1
+    state.save()  # инжект не доехал: mark_told не звали
+
+    state = Seen.load(state.path)
+    again = digest(
+        state, swept([read_stack("beaver-agent", log())], at=NOW + timedelta(hours=3))
+    )
+    assert [e.sample for e in again.new] == [e.sample for e in first.new]
+    assert again.quiet == ()
+
+
+def test_the_state_survives_a_restart(tmp_path):
+    scan = read_stack(
+        "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
+    )
+    state = seen(tmp_path)
+    state.mark_told(digest(state, swept([scan])))
+    state.save()
+
+    state = Seen.load(state.path)
+    assert not state.fresh
+    report = digest(state, swept([scan], at=NOW + timedelta(hours=3)))
+    assert report.new == () and report.quiet == ()
+
+
+def test_an_old_line_still_lying_in_the_tail_is_not_new_again(tmp_path):
+    # Тихий стек держит августовскую строку в хвосте месяцами: она не новая.
+    scan = read_stack(
+        "democalc", log(line("pg-1", "2026-08-22T15:49:51.0", "ERROR: relation"))
+    )
+    state = seen(tmp_path)
+    state.mark_told(digest(state, swept([scan])))
+    state.save()
+
+    state = Seen.load(state.path)
+    assert digest(state, swept([scan], at=NOW + timedelta(hours=3))).new == ()
+
+
+def test_a_trouble_forgotten_by_time_comes_back_as_new(tmp_path):
+    scan = read_stack(
+        "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
+    )
+    state = seen(tmp_path)
+    state.mark_told(digest(state, swept([scan])))
+    late = NOW + KEEP + timedelta(days=1)
+    report = digest(state, swept([scan], at=late))
+    assert len(report.new) == 1
+
+
+def test_a_silent_pass_is_not_worth_the_master(tmp_path):
+    state = seen(tmp_path)
+    assert not digest(state, swept([read_stack("beaver-agent", log())])).worth_telling
+
+
+def test_the_scope_says_how_much_was_looked_at(tmp_path):
+    scans = [
+        read_stack(
+            "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
+        ),
+        StackScan(stack="gitea", error="timeout"),
+    ]
+    report = digest(seen(tmp_path), swept(scans, in_scope=2, fleet=53))
+    head = brief(report).splitlines()[1]
+    assert head.startswith(
+        "Область: dell-home, personal-main - прочитано 1 стеков из 2"
+    )
+    assert "во флоте 53" in head
+    assert "хвост 2000 строк на стек: строки с 20.09 17:40 по 20.09 17:40" in head
+    assert "самый короткий хвост у beaver-agent - только с 20.09 17:40" in head
+
+
+def test_a_tail_shorter_than_the_pause_between_passes_is_said_out_loud(tmp_path):
+    state = seen(tmp_path)
+    state.last_pass = NOW - timedelta(hours=3)
+    scan = read_stack(
+        "beaver-agent",
+        log(line("svc", "2026-09-20T17:59:00.0", "ERROR из последней минуты")),
+    )
+    report = digest(state, swept([scan]))
+    assert report.gaps == ("beaver-agent",)
+    assert "часть строк прошла мимо" in brief(report)
+
+
+def test_a_silent_komodo_says_so_instead_of_reporting_clean(tmp_path):
+    down = Sweep(at=NOW, unreachable="ClientError: connection refused")
+    report = digest(seen(tmp_path), down)
+    assert report.worth_telling
+    text = brief(report)
+    assert "Komodo не ответил" in text
+    assert "Нового в этой области нет" not in text
+
+
+def test_a_lasting_outage_is_told_once(tmp_path):
+    down = Sweep(at=NOW, unreachable="ClientError: connection refused")
+    state = seen(tmp_path)
+    state.mark_told(digest(state, down))
+    assert digest(state, down).trouble == ()
+
+
+async def test_the_sweep_lists_every_page_and_skips_other_peoples_machines():
+    pages = [
+        [a_stack(f"s{i}") for i in range(500)],
+        [a_stack("gitea", "personal-main"), a_stack("veris-runner", "veris-main")],
+    ]
+    calls = []
+    komodo = fleet(
+        {
+            "ListStacks": lambda p: pages[p["page"]],
+            "GetStacksSummary": {"total": 502},
+            "GetStackLog": log(),
+        },
+        calls,
+    )
+    out = await sweep_fleet(komodo, now=NOW)
+    asked = [p["stack"] for r, p in calls if r == "GetStackLog"]
+    assert "veris-runner" not in asked
+    assert "gitea" in asked and "s0" in asked
+    assert out.fleet == 502 and out.in_scope == 501
+    assert out.trouble == ()
+    assert all(p.get("timestamps") for r, p in calls if r == "GetStackLog")
+
+
+async def test_the_sweep_shouts_when_the_list_and_the_summary_disagree():
+    komodo = fleet(
+        {
+            "ListStacks": [a_stack("beaver-agent")],
+            "GetStacksSummary": {"total": 53},
+            "GetStackLog": log(),
+        }
+    )
+    out = await sweep_fleet(komodo, now=NOW)
+    assert out.trouble == (
+        "листание дало 1 стеков, а GetStacksSummary - 53: список неполон",
+    )
+
+
+async def test_a_stack_that_throws_does_not_end_the_sweep():
+    def logs(params):
+        if params["stack"] == "gitea":
+            msg = "connection reset"
+            raise RuntimeError(msg)
+        return log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
+
+    komodo = fleet(
+        {
+            "ListStacks": [a_stack("beaver-agent"), a_stack("gitea", "personal-main")],
+            "GetStacksSummary": {"total": 2},
+            "GetStackLog": logs,
+        }
+    )
+    out = await sweep_fleet(komodo, now=NOW)
+    assert len(out.read) == 1
+    assert out.trouble == ("gitea: RuntimeError: connection reset",)
+
+
+async def test_a_dead_komodo_makes_a_sweep_that_admits_it():
+    class Dead(Komodo):
+        async def read(self, request, params):
+            msg = "connection refused"
+            raise OSError(msg)
+
+    out = await sweep_fleet(Dead(url="u", key="k", secret="s"), now=NOW)
+    assert out.read == ()
+    assert out.unreachable == "OSError: connection refused"
+    assert out.trouble == ("Komodo не ответил: OSError: connection refused",)
+
+
+def test_a_line_without_a_stamp_is_not_a_find():
+    # Голову хвоста Komodo обрезает по месту: полстроки - не новая беда.
+    scan = read_stack(
+        "democalc",
+        log(
+            "49.260 UTC [8772] ERROR:  relation does not exist",
+            line("postgres-1", "2026-09-20T17:40:00.0", "ERROR: настоящая"),
+        ),
+    )
+    assert [h.sample for h in scan.hits] == ["ERROR: настоящая"]
+    assert scan.unparsed == 1 and scan.lines == 1
+
+
+def test_many_unreadable_lines_are_said_out_loud(tmp_path):
+    scan = StackScan(stack="beaver-agent", lines=5, unparsed=40)
+    head = brief(digest(seen(tmp_path), swept([scan]))).splitlines()[1]
+    assert "строк без метки времени пропущено 40" in head
+
+
+def test_an_empty_line_is_a_line_not_a_blind_spot():
+    scan = read_stack("dell-home-dawarich", log("app-1  | 2026-09-17T14:37:38.510Z"))
+    assert scan.unparsed == 0 and scan.lines == 1 and scan.hits == ()
+
+
+def test_a_stack_read_for_the_first_time_does_not_dump_its_past(tmp_path):
+    state = seen(tmp_path)
+    scan = read_stack(
+        "lodge", log(line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас"))
+    )
+    report = digest(state, swept([scan]))
+    assert report.new == ()
+    assert report.baselined == ("lodge",) and report.baseline == 1
+    assert "Впервые прочитаны" in brief(report)
+
+    later = read_stack(
+        "lodge",
+        log(
+            line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас"),
+            line("svc", "2026-09-20T19:00:00.0", "ERROR уже при нас"),
+        ),
+    )
+    after = digest(state, swept([later], at=NOW + timedelta(hours=3)))
+    assert [e.sample for e in after.new] == ["ERROR уже при нас"]
+
+
+def test_a_first_pass_into_a_dead_komodo_does_not_burn_the_baseline(tmp_path):
+    state = seen(tmp_path, fresh=True)
+    down = Sweep(at=NOW, unreachable="OSError: connection refused")
+    report = digest(state, down)
+    assert not report.cold and state.fresh
+    assert report.trouble == ("Komodo не ответил: OSError: connection refused",)
+
+    scan = read_stack(
+        "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
+    )
+    later = digest(state, swept([scan], at=NOW + timedelta(hours=3)))
+    assert later.cold and later.new == ()
+
+
+def test_a_named_secret_in_the_line_does_not_ride_into_the_chat():
+    scan = read_stack(
+        "gitea",
+        log(line("svc", "2026-09-20T17:40:00.0", "ERROR auth failed token=ghp_ZZ123")),
+    )
+    (hit,) = scan.hits
+    assert "ghp_ZZ123" not in hit.sample
+    assert hit.sample.startswith("ERROR auth failed token=")
+
+
+def test_a_bare_blob_stays_readable():
+    # Размен с открытыми глазами: id задач, сессий и коммитов - это сигнал,
+    # и замазывать их ради редкого безымянного ключа дороже, чем оставить.
+    scan = read_stack(
+        "komodo",
+        log(line("svc", "2026-09-20T17:40:00.0", "ERROR task kJ8sdfLKJ23kjhsdf98234x")),
+    )
+    (hit,) = scan.hits
+    assert "kJ8sdfLKJ23kjhsdf98234x" in hit.sample
+
+
+def stamps(day, hour, count, start=0):
+    return [
+        f"2026-09-{day:02d}T{hour:02d}:{m:02d}:00.0"
+        for m in range(start, start + count)
+    ]
+
+
+def burst(stack, times, text="ERROR всё та же беда", service="svc"):
+    """Хвост стека, где одна и та же беда повторилась в перечисленные минуты."""
+    return read_stack(stack, log(*(line(service, t, text) for t in times)))
+
+
+def at(day, hour, minute=0):
+    return datetime(2026, 9, day, hour, minute, tzinfo=UTC)
+
+
+def told(state, report):
+    state.mark_told(report)
+    return report
+
+
+def test_a_continuing_trouble_is_told_once_and_then_keeps_quiet(tmp_path):
+    state = seen(tmp_path)
+    first = told(
+        state, digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, 10)))
+    )
+    assert len(first.new) == 1
+
+    # Хвост перечитан: те же строки плюс свежие - беда идёт, но новостью не стала.
+    later = digest(
+        state,
+        swept([burst("gitea", stamps(20, 9, 3) + stamps(20, 12, 4))], at=at(20, 13)),
+    )
+    assert later.new == () and later.surge == () and later.quiet == ()
+    assert not later.worth_telling
+    assert later.ongoing == ()  # за прошлые сутки её ещё не было
+
+
+def test_the_reread_tail_does_not_inflate_the_count(tmp_path):
+    state = seen(tmp_path)
+    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, 10))))
+    for hour in (13, 16, 19):
+        digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, hour)))
+    (entry,) = state.entries.values()
+    assert entry.count == 3  # три повтора, а не три на каждый из четырёх проходов
+    assert entry.passes == 1
+
+
+def test_a_continuing_trouble_comes_back_in_the_daily_summary(tmp_path):
+    state = seen(tmp_path)
+    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 5))], at=at(20, 10))))
+
+    # Первый проход следующих суток - 01:40, он же несёт сводку.
+    night = digest(
+        state,
+        swept([burst("gitea", stamps(20, 9, 5) + stamps(21, 1, 2))], at=at(21, 1, 40)),
+    )
+    assert night.summary and len(night.ongoing) == 1
+    assert night.new == () and not night.worth_telling
+    text = daily(night, UTC)
+    assert "сводка за 2026-09-20" in text  # числа - про закрывшиеся сутки
+    assert "gitea/svc ошибка, с 20.09 09:00, за сутки 5, первые сутки" in text
+    assert "ERROR" not in text  # образцов строк в сводке нет намеренно
+    told(state, night)
+    state.mark_daily(night)
+
+    # Тот же день, следующий проход: сводку уже отдали.
+    again = digest(
+        state,
+        swept([burst("gitea", stamps(20, 9, 5) + stamps(21, 1, 2))], at=at(21, 4, 40)),
+    )
+    assert again.ongoing and not again.summary
+
+
+def test_the_summary_says_which_way_it_is_going(tmp_path):
+    state = seen(tmp_path)
+    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 2))], at=at(20, 10))))
+    day_two = swept(
+        [burst("gitea", stamps(20, 9, 2) + stamps(21, 9, 8))], at=at(21, 10)
+    )
+    told(state, digest(state, day_two))
+    third = digest(
+        state,
+        swept(
+            [burst("gitea", stamps(20, 9, 2) + stamps(21, 9, 8) + stamps(22, 1, 1))],
+            at=at(22, 1, 40),
+        ),
+    )
+    assert "за сутки 8, растёт (было 2)" in daily(third, UTC)
+
+
+def test_a_surge_of_the_familiar_breaks_the_silence(tmp_path):
+    state = seen(tmp_path)
+    state_report = digest(
+        state, swept([burst("gitea", stamps(20, 9, 1))], at=at(20, 10))
+    )
+    told(state, state_report)
+    for hour, minute in ((13, 12), (16, 15)):
+        digest(
+            state,
+            swept([burst("gitea", stamps(20, hour - 1, 1, minute))], at=at(20, hour)),
+        )
+    (entry,) = state.entries.values()
+    assert entry.passes == 3 and entry.norm == 1.0
+
+    spike = digest(state, swept([burst("gitea", stamps(20, 18, 12))], at=at(20, 19)))
+    (item,) = spike.surge
+    assert item.gain == 12 and item.norm == 1.0
+    assert spike.new == () and spike.worth_telling
+    text = brief(spike)
+    assert "Всплеск знакомого (1)" in text
+    assert "×12 за проход при норме ×1.0" in text
+    assert "Нового в этой области нет" not in text
+
+    # Второй такой же проход внутри выдержки молчит: это уже не новость.
+    told(state, spike)
+    more = digest(state, swept([burst("gitea", stamps(20, 21, 12))], at=at(20, 22)))
+    assert more.surge == () and not more.worth_telling
+
+
+def test_a_small_jump_is_not_a_surge(tmp_path):
+    state = seen(tmp_path)
+    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 1))], at=at(20, 10))))
+    for hour in (13, 16):
+        digest(
+            state,
+            swept([burst("gitea", stamps(20, hour - 1, 1, hour))], at=at(20, hour)),
+        )
+    # Вчетверо больше, но всего четыре штуки: пол в SURGE_MIN держит тишину.
+    small = digest(state, swept([burst("gitea", stamps(20, 18, 4))], at=at(20, 19)))
+    assert small.surge == () and not small.worth_telling
+
+
+def test_a_trouble_gone_silent_is_reported_even_while_its_line_lingers(tmp_path):
+    state = seen(tmp_path)
+    tail = burst("gitea", stamps(20, 9, 2))
+    told(state, digest(state, swept([tail], at=at(20, 10))))
+
+    # Сутки та же строка лежит в хвосте, новых событий нет.
+    assert not digest(state, swept([tail], at=at(20, 22))).quiet
+    stopped = digest(state, swept([tail], at=at(21, 12)))
+    (gone,) = stopped.quiet
+    assert gone.last == at(20, 9, 1)
+    assert "последнее в 20.09 09:01" in brief(stopped)
+
+    told(state, stopped)
+    assert digest(state, swept([tail], at=at(21, 16))).quiet == ()
+
+
+def test_a_trouble_that_returns_after_the_all_clear_is_new_again(tmp_path):
+    state = seen(tmp_path)
+    tail = burst("gitea", stamps(20, 9, 2))
+    told(state, digest(state, swept([tail], at=at(20, 10))))
+    told(state, digest(state, swept([tail], at=at(21, 12))))  # стихла
+
+    back = digest(
+        state,
+        swept([burst("gitea", stamps(20, 9, 2) + stamps(21, 18, 3))], at=at(21, 19)),
+    )
+    assert len(back.new) == 1 and back.worth_telling
+
+
+def test_the_background_of_the_baseline_never_claims_to_have_stopped(tmp_path):
+    # Про фон, принятый в базовую линию, мастер не слышал - «стихло» про него врёт.
+    state = seen(tmp_path, known=())
+    state.fresh = False
+    cold = digest(state, swept([burst("lodge", stamps(20, 9, 2))], at=at(20, 10)))
+    assert cold.new == () and cold.baseline == 1
+    gone = digest(state, swept([read_stack("lodge", log())], at=at(20, 13)))
+    assert gone.quiet == () and not gone.worth_telling
+
+
+def test_the_summary_folds_the_variants_of_one_service_into_one_row(tmp_path):
+    state = seen(tmp_path)
+    # Разные по смыслу строки: отличие в одних цифрах - это одна и та же беда.
+    variants = [
+        burst("gitea", stamps(20, 9, 4), text=f"ERROR {what}")
+        for what in ("push отклонён", "миграция не прошла", "ssh-ключ отвергнут")
+    ]
+    told(state, digest(state, swept(variants, at=at(20, 10))))
+    night = digest(state, swept(variants, at=at(21, 1, 40)))
+    (row,) = grouped(night.ongoing)
+    assert row.kinds == 3 and row.day_before == 12
+    text = daily(night, UTC)
+    assert "Идёт до сих пор: 1 рядов (3 разных строк)" in text
+    assert "gitea/svc ошибка, строк 3 видов, с 20.09 09:00, за сутки 12" in text
+
+
+def test_the_daily_summary_arrives_even_when_all_is_quiet(tmp_path):
+    # Мёртвый джоб и чистый флот должны выглядеть по-разному.
+    state = seen(tmp_path)
+    for hour in (10, 13, 16, 19, 22):
+        digest(state, swept([read_stack("gitea", log())], at=at(20, hour)))
+    night = digest(state, swept([read_stack("gitea", log())], at=at(21, 1, 40)))
+    assert night.summary and night.ongoing == ()
+    assert not night.worth_telling  # обычного инжекта нет, суточный - есть
+    text = daily(night, UTC)
+    assert text.splitlines() == [
+        "Логи флота: сводка за 2026-09-20.",
+        "Проходов за сутки 5 из 8 ожидавшихся; в последнем прочитано 1 стеков "
+        "из 1 в области (dell-home, personal-main).",
+        "Длящегося нет: всё, о чём я говорил, стихло или не повторялось.",
+    ]
+
+
+def test_the_daily_summary_counts_the_refusals(tmp_path):
+    state = seen(tmp_path)
+    digest(state, swept([StackScan(stack="gitea", error="timeout")], at=at(20, 10)))
+    digest(state, Sweep(at=at(20, 13), unreachable="OSError: нет связи"))
+    night = digest(state, swept([read_stack("gitea", log())], at=at(21, 1, 40)))
+    assert (
+        "Отказов доступа за те сутки: чтений стеков 1, проходов без ответа Komodo 1"
+        in daily(night, UTC)
+    )
+
+
+def test_a_slowly_growing_trouble_still_breaks_through(tmp_path):
+    # Варёная лягушка: норма замерла на молодости беды и за ростом не тянется.
+    state = seen(tmp_path)
+    told(state, digest(state, swept([burst("gitea", stamps(20, 8, 1))], at=at(20, 9))))
+    counts = [1, 1, 2, 3, 4, 6, 9, 13]
+    hour = 10
+    seen_times = stamps(20, 8, 1)
+    surged = []
+    for n in counts:
+        seen_times = seen_times + stamps(20, hour - 1, n, start=10)
+        report = digest(state, swept([burst("gitea", seen_times)], at=at(20, hour)))
+        if report.surge:
+            surged.append((hour, report.surge[0].gain, report.surge[0].norm))
+            told(state, report)
+        hour += 1
+    (entry,) = state.entries.values()
+    assert entry.norm == 1.0  # заморожена по первым трём активным проходам
+    assert surged and surged[0][1] >= SURGE_MIN
+    assert surged[0][2] == 1.0
+
+
+def test_the_ceiling_fires_without_any_relative_norm(tmp_path):
+    # Беда, чья собственная норма уже огромна: относительный порог её не возьмёт.
+    state = seen(tmp_path)
+    times = stamps(20, 8, 40)
+    told(state, digest(state, swept([burst("gitea", times)], at=at(20, 9))))
+    for hour in (12, 15):
+        times = times + stamps(20, hour - 1, 40, start=10)
+        digest(state, swept([burst("gitea", times)], at=at(20, hour)))
+    (entry,) = state.entries.values()
+    assert entry.norm == 40.0
+    huge = stamps(20, 17, 55, start=0)
+    spike = digest(state, swept([burst("gitea", times + huge)], at=at(20, 18)))
+    (item,) = spike.surge
+    assert item.gain == 55  # 55 < 4 × 40, взял потолок, а не кратность
+    assert item.gain < SURGE_FACTOR * item.norm
+    assert item.gain >= SURGE_CEILING
+    assert "Всплеск знакомого (1)" in brief(spike)
+
+
+def test_the_journal_is_appended_and_survives_the_passes(tmp_path):
+    # Состояние - снимок, истории в нём нет: пороги пересматривать по журналу.
+    state = seen(tmp_path)
+    path = tmp_path / "логи-флота.jsonl"
+    times = stamps(20, 8, 2)
+    for i, hour in enumerate((10, 13, 16)):
+        times = times + stamps(20, hour - 1, 2, start=20)
+        report = digest(state, swept([burst("gitea", times)], at=at(20, hour)))
+        told(state, report)
+        note(path, record(report, UTC) | {"pass_of_day": i + 1})
+
+    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    assert len(rows) == 3
+    assert [r["pass_of_day"] for r in rows] == [1, 2, 3]
+    assert rows[0]["stacks"] == 1 and rows[0]["in_scope"] == 1
+    assert rows[0]["new"] == 1 and rows[0]["misses"] == 0
+    (trouble,) = rows[2]["troubles"]
+    assert trouble["where"] == "gitea/svc" and trouble["gain"] == 2
+    assert trouble["total"] == 8 and trouble["day"] == 8
+
+
+def test_the_journal_turns_over_instead_of_growing_forever(tmp_path):
+    path = tmp_path / "логи-флота.jsonl"
+    for i in range(4):
+        note(path, {"n": i, "padding": "x" * 200}, max_bytes=300)
+    fresh = [json.loads(ln) for ln in path.read_text(encoding="utf-8").splitlines()]
+    turned = [
+        json.loads(ln)
+        for ln in path.with_suffix(".jsonl.1").read_text(encoding="utf-8").splitlines()
+    ]
+    assert [r["n"] for r in turned] == [0, 1]  # старое уехало в .1, не пропало
+    assert [r["n"] for r in fresh] == [2, 3]
+
+
+def test_a_storm_of_new_kinds_is_folded_into_one_loud_line(tmp_path):
+    state = seen(tmp_path)
+    flood = [
+        burst("gitea", stamps(20, 9, 1, start=i), text=f"ERROR сбой {chr(1072 + i)}щ")
+        for i in range(MUTE_KINDS + 5)
+    ]
+    report = digest(state, swept(flood, at=at(20, 10)))
+    (storm,) = report.storms
+    assert storm.kinds == MUTE_KINDS + 5 and storm.shown == MUTE_SHOW
+    assert len(report.new) == MUTE_SHOW  # остальное проглочено, но посчитано
+    text = brief(report)
+    assert "глушу" in text
+    assert f"Стек gitea: {MUTE_KINDS + 5} новых ВИДОВ строк за проход" in text
+    assert report.worth_telling
+
+    # Пока не сказали - повторяем; сказали - в суточной сводке.
+    told(state, report)
+    assert state.mutes["gitea"].told
+    assert state.mutes["gitea"].kinds == 5 + MUTE_KINDS - MUTE_SHOW
+
+
+def test_the_muted_stack_is_named_once_a_day_until_it_calms_down(tmp_path):
+    state = seen(tmp_path)
+    flood = [
+        burst("gitea", stamps(20, 9, 1, start=i), text=f"ERROR сбой {chr(1072 + i)}щ")
+        for i in range(MUTE_KINDS + 5)
+    ]
+    told(state, digest(state, swept(flood, at=at(20, 10))))
+    night = digest(state, swept(flood, at=at(21, 1, 40)))
+    assert "Приглушён gitea" in daily(night, UTC)
+    assert "глушу с 20.09 10:00" in daily(night, UTC)
+
+    calm = digest(state, swept([burst("gitea", stamps(21, 9, 2))], at=at(21, 10)))
+    assert calm.unmuted == ("gitea",) and "глушение снял: gitea" in brief(calm)
+    assert "gitea" not in state.mutes
+
+
+def test_the_first_sighting_is_never_a_surge(tmp_path):
+    # В хвосте лежит суточный запас: первый счёт - про длину хвоста, не про беду.
+    state = seen(tmp_path, fresh=True)
+    loud = digest(state, swept([burst("gitea", stamps(20, 9, 55))], at=at(20, 10)))
+    assert loud.surge == () and loud.cold and loud.baseline == 1