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

beaver_agent/hands/komodo_logs.py

@@ -0,0 +1,541 @@
+"""Логи флота периодами: что в хвостах похоже на беду и чего там раньше не было.
+
+Ходит и читает джоб `jobs/logs.py`, судит этот модуль. Правило одно: мастера
+дёргает только НОВОЕ. То, что молотит в логе третьи сутки, уже сказано один
+раз и больше не показывается; показывается то, чего в прошлом проходе не было,
+и обратный переход «было → нет» - с временем последнего события, не с временем
+прохода (логи берутся с `--timestamps`, время в строке настоящее).
+
+Показание всегда несёт свою область: сколько стеков в области, сколько из них
+прочиталось, какой отрезок логов попал в хвост. Без этого «всё чисто»
+неотличимо от «половину не посмотрел», поэтому чего не смог - тоже в инжекте.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta, tzinfo
+from typing import TYPE_CHECKING, Any
+
+from beaver_agent.hands.komodo import ANSI
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+__all__ = [
+    "SCOPE",
+    "SIGNALS",
+    "TAIL",
+    "Entry",
+    "Report",
+    "Seen",
+    "StackScan",
+    "Sweep",
+    "brief",
+    "digest",
+    "read_stack",
+]
+
+SCOPE = ("dell-home", "personal-main")
+"""Свои серверы. Машины заказчиков (md-cars-*, veris-*, ms-agents-*) не наши:
+там чужой прод и чужие секреты, и читать их логи в свой контекст - нельзя."""
+
+TAIL = 500
+"""Строк на стек. Болтливый стек столько строк проживает за минуты - об этом
+в инжекте говорится отдельно, а не выдаётся за просмотренный промежуток."""
+
+SIGNALS: tuple[tuple[str, str], ...] = (
+    ("критично", r"\bCRITICAL\b|\bFATAL\b|\bEMERG\b"),
+    ("паника", r"\bpanic:|\bpanicked at\b|\bfatal error:"),
+    ("oom", r"\bOOM\b|OOMKilled|[Oo]ut of memory|oom[-_]kill"),
+    ("traceback", r"Traceback \(most recent call last\)|\bUnhandled exception\b"),
+    ("падение", r"exited with code (?!0\b)\d+|\bexit status [1-9]|Exited \([1-9]\d*\)"),
+    ("рестарт-цикл", r"[Rr]estarting \(\d+\)|back-?off.{0,20}restart|CrashLoopBackOff"),
+    ("healthcheck", r"health-?\s?check (?:failed|error)|\bunhealthy\b"),
+    ("ошибка", r"\bERROR\b|\bERR!\b"),
+)
+"""Что считается бедой: метка и выражение. Правится одним местом; порядок -
+приоритет, строка получает первую подошедшую метку."""
+
+MATCHERS = tuple((label, re.compile(pattern)) for label, pattern in SIGNALS)
+
+SELF_MARK = "Логи флота"
+"""Свой же инжект видно в stdout gateway: ловить себя по кругу не надо."""
+
+LINE_RE = re.compile(r"^(?P<service>[^\s|]+)\s*\|\s?(?P<rest>.*)$")
+STAMP_RE = re.compile(
+    r"^(?P<ts>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)(?:\s(?P<text>.*))?$"
+)
+"""Пустая строка тоже со штампом (`сервис | время` и ничего) - это не мусор."""
+DIGITS_RE = re.compile(r"\b[0-9a-f]{7,}\b|\d+")
+"""Числа и хеши в строке - шум: без них «то же самое» узнаётся между проходами."""
+SECRET_RE = re.compile(
+    r"(?i)\b(?:token|secret|api[-_ ]?key|password|passwd|authorization)\b\s*[=:]\s*\S+"
+    r"|\b[A-Za-z0-9_\-]{32,}\b"
+)
+"""Инжект уезжает в чат и в vault: похожее на ключ в показании не нужно."""
+
+SAMPLE_CHARS = 180
+MAX_NEW = 25
+MAX_QUIET = 10
+MAX_TROUBLE = 10
+GAP_NAMES = 3
+UNPARSED_LOUD = 10
+"""Столько неразобранных строк на проход - уже не обрезанная голова, а дыра."""
+KEEP = timedelta(days=14)
+"""Сколько помнить беду, которой больше не видно: дальше пусть будет новой."""
+KEY_CHARS = 12
+STATE_VERSION = 1
+EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
+
+
+@dataclass(frozen=True, slots=True)
+class Hit:
+    """Одна беда в хвосте одного стека, уже слитая по повторам."""
+
+    key: str
+    stack: str
+    service: str
+    signal: str
+    sample: str
+    at: datetime | None
+    count: int
+
+
+@dataclass(frozen=True, slots=True)
+class StackScan:
+    """Что дал один стек: находки, отрезок его хвоста или причина отказа."""
+
+    stack: str
+    hits: tuple[Hit, ...] = ()
+    first: datetime | None = None
+    last: datetime | None = None
+    lines: int = 0
+    unparsed: int = 0
+    """Строки без `сервис | время`: обрезанная голова хвоста и прочий мусор."""
+    error: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.error is None
+
+
+@dataclass(frozen=True, slots=True)
+class Sweep:
+    """Проход целиком - вместе с тем, какую область он покрыл."""
+
+    at: datetime
+    scans: tuple[StackScan, ...] = ()
+    in_scope: int = 0
+    fleet: int = 0
+    summary: int | None = None
+    tail: int = TAIL
+    scope: tuple[str, ...] = SCOPE
+    unreachable: str | None = None
+    """Komodo не ответил вовсе - ни одного стека, и это не «чисто»."""
+
+    @property
+    def read(self) -> tuple[StackScan, ...]:
+        return tuple(s for s in self.scans if s.ok)
+
+    @property
+    def trouble(self) -> tuple[str, ...]:
+        if self.unreachable:
+            return (f"Komodo не ответил: {self.unreachable}",)
+        out = [f"{s.stack}: {s.error}" for s in self.scans if not s.ok]
+        if self.summary is not None and self.summary != self.fleet:
+            out.append(
+                f"листание дало {self.fleet} стеков, а GetStacksSummary - "
+                f"{self.summary}: список неполон"
+            )
+        return tuple(out)
+
+    @property
+    def span(self) -> tuple[datetime | None, datetime | None]:
+        """Отрезок, который реально попал в хвосты всех прочитанных стеков."""
+        firsts = [s.first for s in self.read if s.first]
+        lasts = [s.last for s in self.read if s.last]
+        return (min(firsts) if firsts else None, max(lasts) if lasts else None)
+
+    @property
+    def shortest(self) -> StackScan | None:
+        """Стек с самым коротким хвостом: он и ограничивает, что мы видели."""
+        stamped = [s for s in self.read if s.first]
+        if not stamped:
+            return None
+        return max(stamped, key=lambda s: s.first or EPOCH)
+
+    def gaps(self, since: datetime | None) -> tuple[str, ...]:
+        """Стеки, чей хвост начался позже прошлого прохода: там дыра."""
+        if since is None:
+            return ()
+        return tuple(s.stack for s in self.read if s.first and s.first > since)
+
+
+@dataclass(slots=True)
+class Entry:
+    """Беда, которую уже видели: когда впервые, когда последний раз, сколько раз."""
+
+    key: str
+    stack: str
+    service: str
+    signal: str
+    sample: str
+    first: datetime | None = None
+    last: datetime | None = None
+    """Время события из строки лога, а не время прохода."""
+    looked: datetime | None = None
+    """Когда её последний раз видели в хвосте: по этому и забываем."""
+    count: int = 0
+    told: bool = False
+    """Сказали ли о ней мастеру. Инжект не доехал - не сказали, повторим."""
+
+    @property
+    def where(self) -> str:
+        return f"{self.stack}/{self.service}"
+
+
+@dataclass(frozen=True, slots=True)
+class Report:
+    """Что из прохода стоит мастера."""
+
+    sweep: Sweep
+    new: tuple[Entry, ...] = ()
+    quiet: tuple[Entry, ...] = ()
+    trouble: tuple[str, ...] = ()
+    gaps: tuple[str, ...] = ()
+    cold: bool = False
+    """Первый проход: состояния не было, и вся область легла в базовую линию."""
+    baseline: int = 0
+    """Сколько бед легло в базовую линию в этом проходе."""
+    baselined: tuple[str, ...] = ()
+    """Стеки, впервые прочитанные в этом проходе: их прошлое не показывается."""
+
+    @property
+    def worth_telling(self) -> bool:
+        return bool(self.new or self.quiet or self.trouble or self.cold)
+
+
+@dataclass(slots=True)
+class Seen:
+    """Состояние между проходами: файл на диске, переживает передеплой."""
+
+    path: Path
+    entries: dict[str, Entry] = field(default_factory=dict)
+    baselined: set[str] = field(default_factory=set)
+    """Стеки, чьё прошлое уже принято за данность: их старое не «новое»."""
+    trouble: tuple[str, ...] = ()
+    """О каких отказах уже сказали: Komodo лежит - скажем один раз, не каждые 3 ч."""
+    last_pass: datetime | None = None
+    fresh: bool = True
+    """Файла не было: базовую линию только снимаем, стену старых бед не шлём."""
+
+    @classmethod
+    def load(cls, path: Path) -> Seen:
+        try:
+            raw = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            return cls(path=path)
+        if raw.get("version") != STATE_VERSION:
+            return cls(path=path)
+        entries = {
+            key: Entry(
+                key=key,
+                stack=row["stack"],
+                service=row["service"],
+                signal=row["signal"],
+                sample=row["sample"],
+                first=_parse(row.get("first")),
+                last=_parse(row.get("last")),
+                looked=_parse(row.get("looked")),
+                count=int(row.get("count", 0)),
+                told=bool(row.get("told")),
+            )
+            for key, row in raw.get("entries", {}).items()
+        }
+        return cls(
+            path=path,
+            entries=entries,
+            baselined=set(raw.get("baselined", ())),
+            trouble=tuple(raw.get("trouble", ())),
+            last_pass=_parse(raw.get("last_pass")),
+            fresh=False,
+        )
+
+    def save(self) -> None:
+        body = {
+            "version": STATE_VERSION,
+            "last_pass": _iso(self.last_pass),
+            "baselined": sorted(self.baselined),
+            "trouble": list(self.trouble),
+            "entries": {
+                key: {
+                    "stack": e.stack,
+                    "service": e.service,
+                    "signal": e.signal,
+                    "sample": e.sample,
+                    "first": _iso(e.first),
+                    "last": _iso(e.last),
+                    "looked": _iso(e.looked),
+                    "count": e.count,
+                    "told": e.told,
+                }
+                for key, e in self.entries.items()
+            },
+        }
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = self.path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(body, ensure_ascii=False), encoding="utf-8")
+        tmp.replace(self.path)
+
+    def mark_told(self, report: Report) -> None:
+        """Инжект доехал: новое больше не новое, стихшее забывается."""
+        for entry in report.new:
+            entry.told = True
+        for entry in report.quiet:
+            self.entries.pop(entry.key, None)
+        self.trouble = report.trouble
+
+
+def _iso(at: datetime | None) -> str | None:
+    return at.isoformat() if at else None
+
+
+def _parse(raw: Any) -> datetime | None:
+    if not isinstance(raw, str):
+        return None
+    try:
+        at = datetime.fromisoformat(raw)
+    except ValueError:
+        return None
+    return at if at.tzinfo else at.replace(tzinfo=UTC)
+
+
+def _mask(text: str) -> str:
+    """Замазать похожее на секрет: строка лога едет мастеру как есть."""
+    return SECRET_RE.sub("<скрыто>", text)
+
+
+def _normal(text: str) -> str:
+    """Строка без чисел, хешей и адресов: по ней узнаётся «та же самая беда»."""
+    return DIGITS_RE.sub("#", ANSI.sub("", text)).strip()
+
+
+def _key(stack: str, service: str, text: str) -> str:
+    raw = f"{stack}\0{service}\0{_normal(text)}".encode()
+    return hashlib.blake2s(raw, digest_size=KEY_CHARS // 2).hexdigest()
+
+
+def signal_of(text: str) -> str | None:
+    for label, matcher in MATCHERS:
+        if matcher.search(text):
+            return label
+    return None
+
+
+def read_stack(stack: str, log: dict[str, Any]) -> StackScan:
+    """Ответ GetStackLog → находки стека; ошибка Komodo остаётся ошибкой."""
+    if not isinstance(log, dict) or "error" in log:
+        error = log.get("error") if isinstance(log, dict) else "ответ не разобран"
+        return StackScan(stack=stack, error=str(error)[:200])
+    text = (log.get("stdout") or "") + "\n" + (log.get("stderr") or "")
+    found: dict[str, Hit] = {}
+    stamps: list[datetime] = []
+    lines = 0
+    unparsed = 0
+    for raw in text.splitlines():
+        line = ANSI.sub("", raw).strip()
+        if not line:
+            continue
+        match = LINE_RE.match(line)
+        stamped = STAMP_RE.match(match.group("rest")) if match else None
+        if match is None or stamped is None:
+            # Голову хвоста Komodo обрезает по месту, и от прохода к проходу она
+            # разная: недоразобранная строка - не находка, иначе каждый раз «ново».
+            unparsed += 1
+            continue
+        lines += 1
+        service = match.group("service")
+        at = datetime.fromisoformat(stamped.group("ts")).astimezone(UTC)
+        stamps.append(at)
+        rest = stamped.group("text") or ""
+        if SELF_MARK in rest:
+            continue
+        signal = signal_of(rest)
+        if signal is None:
+            continue
+        key = _key(stack, service, rest)
+        before = found.get(key)
+        found[key] = Hit(
+            key=key,
+            stack=stack,
+            service=service,
+            signal=signal,
+            sample=(before.sample if before else _mask(rest)[:SAMPLE_CHARS]),
+            at=at or (before.at if before else None),
+            count=(before.count if before else 0) + 1,
+        )
+    return StackScan(
+        stack=stack,
+        hits=tuple(found.values()),
+        first=min(stamps) if stamps else None,
+        last=max(stamps) if stamps else None,
+        lines=lines,
+        unparsed=unparsed,
+    )
+
+
+def _forget(seen: Seen, now: datetime) -> None:
+    """Беду, которой давно не видно в хвостах, забываем: вернётся - будет новой.
+
+    Считаем по времени прохода, в котором её видели, а не по времени события:
+    строка от прошлого месяца может спокойно лежать в хвосте тихого стека, и
+    забывать её - значит каждый проход объявлять её новой.
+    """
+    for key, entry in list(seen.entries.items()):
+        if entry.told and entry.looked is not None and now - entry.looked > KEEP:
+            del seen.entries[key]
+
+
+def digest(seen: Seen, sweep: Sweep) -> Report:
+    """Сравнить проход с прошлым: что новое, что стихло, о чём молчать."""
+    gaps = sweep.gaps(seen.last_pass)
+    _forget(seen, sweep.at)
+    looked = {scan.stack for scan in sweep.read}
+    newcomers = looked - seen.baselined
+    hits = {hit.key: hit for scan in sweep.read for hit in scan.hits}
+    baseline = 0
+    for key, hit in hits.items():
+        entry = seen.entries.get(key)
+        if entry is None:
+            entry = Entry(
+                key=key,
+                stack=hit.stack,
+                service=hit.service,
+                signal=hit.signal,
+                sample=hit.sample,
+                first=hit.at or sweep.at,
+                looked=sweep.at,
+            )
+            seen.entries[key] = entry
+            if hit.stack in newcomers:
+                # Стек читаем впервые: в хвосте лежит всё его прошлое, и вываливать
+                # его как «новое» - врать. Принимаем за данность, дальше - только рост.
+                entry.told = True
+                baseline += 1
+        entry.last = hit.at or entry.last or sweep.at
+        entry.looked = sweep.at
+        entry.count += hit.count
+    seen.baselined |= looked
+    # Стихло - только у стеков, которые прочитались: чего не видели, того не судим.
+    quiet = tuple(
+        e
+        for e in seen.entries.values()
+        if e.told and e.stack in looked and e.key not in hits
+    )
+    seen.last_pass = sweep.at
+    # Холодный старт без единого прочитанного стека базой не считается: иначе
+    # первый удачный проход вывалит всю область как «новое».
+    cold = seen.fresh and bool(sweep.read)
+    seen.fresh = seen.fresh and not sweep.read
+    fresh = tuple(e for e in seen.entries.values() if not e.told)
+    trouble = sweep.trouble if sweep.trouble != seen.trouble else ()
+    return Report(
+        sweep=sweep,
+        new=tuple(sorted(fresh, key=lambda e: (e.stack, e.service))),
+        quiet=tuple(sorted(quiet, key=lambda e: (e.stack, e.service))),
+        trouble=trouble,
+        gaps=gaps,
+        cold=cold,
+        baseline=baseline,
+        baselined=tuple(sorted(newcomers)),
+    )
+
+
+def _when(at: datetime | None, tz: tzinfo) -> str:
+    return at.astimezone(tz).strftime("%d.%m %H:%M") if at else "время не в строке"
+
+
+def _scope_line(sweep: Sweep, gaps: tuple[str, ...], tz: tzinfo) -> str:
+    since, until = sweep.span
+    where = ", ".join(sweep.scope)
+    if sweep.unreachable:
+        return f"Область: {where} - не прочитано ничего, списка стеков нет."
+    out = [
+        (
+            f"Область: {where} - прочитано {len(sweep.read)} стеков из "
+            f"{sweep.in_scope} в области (во флоте {sweep.fleet})"
+        ),
+        (
+            f"хвост {sweep.tail} строк на стек: строки с {_when(since, tz)} "
+            f"по {_when(until, tz)}"
+        ),
+    ]
+    short = sweep.shortest
+    if short is not None:
+        out.append(
+            f"самый короткий хвост у {short.stack} - только с {_when(short.first, tz)}"
+        )
+    blind = sum(s.unparsed for s in sweep.read)
+    if blind > UNPARSED_LOUD:
+        out.append(f"строк без метки времени пропущено {blind} - их не смотрел")
+    if gaps:
+        names = ", ".join(gaps[:GAP_NAMES])
+        more = f" и ещё {len(gaps) - GAP_NAMES}" if len(gaps) > GAP_NAMES else ""
+        out.append(
+            f"у {len(gaps)} стеков хвост начался позже прошлого прохода "
+            f"({names}{more}) - между проходами часть строк прошла мимо"
+        )
+    return "; ".join(out) + "."
+
+
+def _entry_line(entry: Entry, tz: tzinfo, *, new: bool) -> str:
+    when = _when(entry.last, tz)
+    times = f" ×{entry.count}" if entry.count > 1 else ""
+    head = f"- {entry.where} {when}{times} {entry.signal}: "
+    if not new:
+        head = f"- {entry.where} {entry.signal}, последнее в {when}: "
+    return head + entry.sample
+
+
+def brief(report: Report, tz: tzinfo = UTC) -> str:
+    """Инжект: что нового, где, сколько раз - и всегда область просмотра."""
+    sweep = report.sweep
+    lines = [f"{SELF_MARK}: проход {sweep.at.astimezone(tz):%d.%m %H:%M}."]
+    lines.append(_scope_line(sweep, report.gaps, tz))
+    if report.cold:
+        lines.append(
+            f"Первый проход: снял базовую линию, в хвостах {report.baseline} "
+            "таких строк - дальше дёргаю только тем, чего в прошлом проходе не было."
+        )
+    elif report.baselined:
+        names = ", ".join(report.baselined[:GAP_NAMES])
+        more = (
+            f" и ещё {len(report.baselined) - GAP_NAMES}"
+            if len(report.baselined) > GAP_NAMES
+            else ""
+        )
+        lines.append(
+            f"Впервые прочитаны ({report.baseline} строк в базу): {names}{more} - "
+            "их прошлое принял за данность, показываю только то, что появится дальше."
+        )
+    if report.new:
+        lines.append(f"Новое ({len(report.new)}):")
+        lines.extend(_entry_line(e, tz, new=True) for e in report.new[:MAX_NEW])
+        if len(report.new) > MAX_NEW:
+            lines.append(f"…и ещё {len(report.new) - MAX_NEW} - смотри komodo(logs).")
+    elif not report.cold and sweep.read:
+        lines.append("Нового в этой области нет.")
+    if report.quiet:
+        lines.append(f"Стихло ({len(report.quiet)}), время - события, не прохода:")
+        lines.extend(_entry_line(e, tz, new=False) for e in report.quiet[:MAX_QUIET])
+        if len(report.quiet) > MAX_QUIET:
+            lines.append(f"…и ещё {len(report.quiet) - MAX_QUIET}.")
+    if report.trouble:
+        lines.append(f"Не смог посмотреть ({len(report.trouble)}):")
+        lines.extend(f"- {t}" for t in report.trouble[:MAX_TROUBLE])
+    lines.append("Это только чтение: чинить или забить - решаешь ты.")
+    return "\n".join(lines)

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),
+    )

beaver_agent/vault.py

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

tests/test_komodo_logs.py

@@ -0,0 +1,465 @@
+from datetime import UTC, datetime, timedelta
+
+from beaver_agent.hands.komodo import Komodo
+from beaver_agent.hands.komodo_logs import (
+    KEEP,
+    SELF_MARK,
+    Seen,
+    StackScan,
+    Sweep,
+    brief,
+    digest,
+    read_stack,
+)
+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 "хвост 500 строк на стек: строки с 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_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 for kJ8sdfLKJ23kjhsdf98234hkjsdf7823x",
+            )
+        ),
+    )
+    (hit,) = scan.hits
+    assert "ghp_ZZ123" not in hit.sample
+    assert "kJ8sdfLKJ23kjhsdf98234hkjsdf7823x" not in hit.sample
+    assert hit.sample.startswith("ERROR auth failed")