"""Раз в 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), )