feat(jobs,hands): сканер логов флота будит только новым и несёт свою область

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,9 @@ 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 чтений за проход), реже - хвост уезжает совсем.
+    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,124 @@
+"""Раз в 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,
+    digest,
+    read_stack,
+)
+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"
+"""Состояние прохода: что уже видели и о чём уже сказали."""
+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:
+    """Проход по логам: состояние на диске, инжект - только на новом."""
+    if KOMODO is None:
+        _log.info("логи: руки komodo нет, проход пропущен")
+        return
+    seen = Seen.load(STATE_FILE)
+    report = digest(seen, await sweep(KOMODO))
+    if not report.worth_telling:
+        seen.save()
+        _log.info(
+            "логи: нового нет, стеков прочитано %s из %s",
+            len(report.sweep.read),
+            report.sweep.in_scope,
+        )
+        return
+    if await run.inject_master(brief(report, ZONE), urgency="normal", origin="логи"):
+        seen.mark_told(report)
+    seen.save()
+    _log.info(
+        "логи: новое %s, стихло %s, не прочитано %s",
+        len(report.new),
+        len(report.quiet),
+        len(report.trouble),
+    )