feat(hands,jobs): длящаяся беда не пропадает - суточная сводка и всплеск против своей нормы

beaver_agent/hands/komodo_logs.py

@@ -6,6 +6,12 @@
 и обратный переход «было → нет» - с временем последнего события, не с временем
 прохода (логи берутся с `--timestamps`, время в строке настоящее).
 
+Молчать - не то же самое, что забыть. Беда, идущая неделю, после первого
+доклада не исчезает: она переходит в длящиеся и раз в сутки идёт отдельной
+короткой сводкой (имена и числа, без строк), а если за проход её стало в разы
+больше собственной нормы - снова дёргает сразу. Затихание наоборот: не тревога,
+а строка «стихло» с временем последнего события.
+
 Показанную строку лога чистит общий `security.redact` - тот же, что стоит на
 логах и на выдаче MCP: правила в одном месте, и он знает литеральные значения
 секретов из env. Голые блобы без имени рядом он не трогает намеренно - id
@@ -37,12 +43,16 @@ __all__ = [
     "SIGNALS",
     "TAIL",
     "Entry",
+    "Ongoing",
     "Report",
     "Seen",
     "StackScan",
+    "Surge",
     "Sweep",
     "brief",
+    "daily",
     "digest",
+    "grouped",
     "read_stack",
 ]
 
@@ -94,10 +104,33 @@ MAX_TROUBLE = 10
 GAP_NAMES = 3
 UNPARSED_LOUD = 10
 """Столько неразобранных строк на проход - уже не обрезанная голова, а дыра."""
+MAX_DAILY = 20
 KEEP = timedelta(days=14)
 """Сколько помнить беду, которой больше не видно: дальше пусть будет новой."""
+
+QUIET_AFTER = timedelta(hours=24)
+"""Сутки без единого нового события - беда считается стихшей, даже если её
+строка ещё лежит в хвосте. Иначе на тихом стеке «стихло» ждало бы неделями,
+пока строку не вытеснит."""
+
+SURGE_FACTOR = 4.0
+"""Во сколько раз проход должен перебить собственную норму беды, чтобы снова
+стоить тревоги. Проходов восемь в сутки, и счёт за проход сам по себе гуляет
+вдвое (деплой, ночной крон, перезапуск) - вчетверо это уже не дрожание.
+Считается по среднему за прошлые активные проходы, то есть у каждой беды
+норма своя; хвост ограничен, поэтому взрыв упирается в TAIL и видно скорее
+«стало на порядок», чем точную кратность."""
+SURGE_MIN = 10
+"""Пол в штуках: без него «было одно, стало четыре» будило бы каждый раз."""
+SURGE_PASSES = 3
+"""Пока беду видели меньше трёх раз, нормы у неё нет и сравнивать не с чем."""
+SURGE_HOLD = timedelta(hours=12)
+"""Одна беда - не чаще раза в полсуток, даже если растёт ступенями."""
+TREND_UP, TREND_DOWN = 1.5, 0.67
+"""Полтора раза в сутках - уже тренд, между ними - то же самое."""
+
 KEY_CHARS = 12
-STATE_VERSION = 1
+STATE_VERSION = 2
 EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
 
 
@@ -112,6 +145,15 @@ class Hit:
     sample: str
     at: datetime | None
     count: int
+    times: tuple[datetime, ...] = ()
+    """Время каждого повтора в хвосте: хвост перечитывается каждый проход, и
+    без этого одни и те же строки считались бы заново по восемь раз в сутки."""
+
+    def since(self, after: datetime | None) -> int:
+        """Сколько повторов случилось позже ``after`` - то есть новых для нас."""
+        if after is None:
+            return self.count
+        return sum(1 for t in self.times if t > after)
 
 
 @dataclass(frozen=True, slots=True)
@@ -199,13 +241,68 @@ class Entry:
     looked: datetime | None = None
     """Когда её последний раз видели в хвосте: по этому и забываем."""
     count: int = 0
+    """Всего повторов, посчитанных приростом, а не перечитыванием хвоста."""
+    passes: int = 0
+    """В скольких проходах она прирастала: по ним и считается её норма."""
+    day_count: int = 0
+    day_before: int = 0
+    day_earlier: int = 0
+    """Повторы за текущие сутки и за двое прошлых - из них сводка и тренд."""
     told: bool = False
     """Сказали ли о ней мастеру. Инжект не доехал - не сказали, повторим."""
+    announced: bool = False
+    """Про неё говорили именно как про новую, а не приняли в базовую линию:
+    «стихло» бывает только про то, о чём мастер слышал."""
+    quieted: datetime | None = None
+    """Когда доложили, что стихла. Появится новое событие - снова новость."""
+    surged: datetime | None = None
+    """Когда последний раз докладывали о всплеске: чаще раза в полсуток - нет."""
 
     @property
     def where(self) -> str:
         return f"{self.stack}/{self.service}"
 
+    @property
+    def norm(self) -> float:
+        """Своя норма: сколько повторов у неё бывает за активный проход."""
+        return self.count / self.passes if self.passes else 0.0
+
+
+@dataclass(frozen=True, slots=True)
+class Ongoing:
+    """Длящееся одним рядом: стек, сервис и сигнал, а не отдельная строка.
+
+    Один сервис сыплет десяток похожих строк (у caddy их семь видов), и в
+    сводке каждая своей строкой - та же стена, от которой сводку и делали.
+    """
+
+    where: str
+    signal: str
+    first: datetime | None
+    day_before: int
+    day_earlier: int
+    kinds: int
+    """Сколько разных строк слилось в этот ряд."""
+
+    def trend(self) -> str:
+        if not self.day_earlier:
+            return "первые сутки"
+        ratio = self.day_before / self.day_earlier
+        if ratio >= TREND_UP:
+            return f"растёт (было {self.day_earlier})"
+        if ratio <= TREND_DOWN:
+            return f"падает (было {self.day_earlier})"
+        return f"ровно (было {self.day_earlier})"
+
+
+@dataclass(frozen=True, slots=True)
+class Surge:
+    """Знакомая беда, которой за проход стало в разы больше своей нормы."""
+
+    entry: Entry
+    gain: int
+    norm: float
+
 
 @dataclass(frozen=True, slots=True)
 class Report:
@@ -213,7 +310,14 @@ class Report:
 
     sweep: Sweep
     new: tuple[Entry, ...] = ()
+    surge: tuple[Surge, ...] = ()
     quiet: tuple[Entry, ...] = ()
+    ongoing: tuple[Entry, ...] = ()
+    """Длящееся: молчит в обычном проходе, идёт в суточную сводку."""
+    summary: bool = False
+    """Сутки закрылись и сводку по длящемуся ещё не отдавали."""
+    day: str = ""
+    """Сутки, за которые считается сводка (те, что только что закрылись)."""
     trouble: tuple[str, ...] = ()
     gaps: tuple[str, ...] = ()
     cold: bool = False
@@ -225,7 +329,8 @@ class Report:
 
     @property
     def worth_telling(self) -> bool:
-        return bool(self.new or self.quiet or self.trouble or self.cold)
+        """Суточная сводка сюда не входит: она едет своим инжектом."""
+        return bool(self.new or self.surge or self.quiet or self.trouble or self.cold)
 
 
 @dataclass(slots=True)
@@ -239,6 +344,10 @@ class Seen:
     trouble: tuple[str, ...] = ()
     """О каких отказах уже сказали: Komodo лежит - скажем один раз, не каждые 3 ч."""
     last_pass: datetime | None = None
+    day: str | None = None
+    """Локальные сутки, которым принадлежат суточные счётчики записей."""
+    summary_day: str | None = None
+    """Сутки, за которые сводка по длящемуся уже доехала до мастера."""
     fresh: bool = True
     """Файла не было: базовую линию только снимаем, стену старых бед не шлём."""
 
@@ -261,7 +370,14 @@ class Seen:
                 last=_parse(row.get("last")),
                 looked=_parse(row.get("looked")),
                 count=int(row.get("count", 0)),
+                passes=int(row.get("passes", 0)),
+                day_count=int(row.get("day_count", 0)),
+                day_before=int(row.get("day_before", 0)),
+                day_earlier=int(row.get("day_earlier", 0)),
                 told=bool(row.get("told")),
+                announced=bool(row.get("announced")),
+                quieted=_parse(row.get("quieted")),
+                surged=_parse(row.get("surged")),
             )
             for key, row in raw.get("entries", {}).items()
         }
@@ -271,6 +387,8 @@ class Seen:
             baselined=set(raw.get("baselined", ())),
             trouble=tuple(raw.get("trouble", ())),
             last_pass=_parse(raw.get("last_pass")),
+            day=raw.get("day"),
+            summary_day=raw.get("summary_day"),
             fresh=False,
         )
 
@@ -278,6 +396,8 @@ class Seen:
         body = {
             "version": STATE_VERSION,
             "last_pass": _iso(self.last_pass),
+            "day": self.day,
+            "summary_day": self.summary_day,
             "baselined": sorted(self.baselined),
             "trouble": list(self.trouble),
             "entries": {
@@ -290,7 +410,14 @@ class Seen:
                     "last": _iso(e.last),
                     "looked": _iso(e.looked),
                     "count": e.count,
+                    "passes": e.passes,
+                    "day_count": e.day_count,
+                    "day_before": e.day_before,
+                    "day_earlier": e.day_earlier,
                     "told": e.told,
+                    "announced": e.announced,
+                    "quieted": _iso(e.quieted),
+                    "surged": _iso(e.surged),
                 }
                 for key, e in self.entries.items()
             },
@@ -301,13 +428,24 @@ class Seen:
         tmp.replace(self.path)
 
     def mark_told(self, report: Report) -> None:
-        """Инжект доехал: новое больше не новое, стихшее забывается."""
+        """Инжект доехал: новое сказано, стихшее помечено, всплеск выдержан.
+
+        Стихшее не выбрасывается, а помечается: его строка может ещё лежать в
+        хвосте, и выброшенная запись завтра же вернулась бы как «новая».
+        """
         for entry in report.new:
             entry.told = True
+            entry.announced = True
+        for item in report.surge:
+            item.entry.surged = report.sweep.at
         for entry in report.quiet:
-            self.entries.pop(entry.key, None)
+            entry.quieted = report.sweep.at
         self.trouble = report.trouble
 
+    def mark_daily(self, report: Report) -> None:
+        """Сводка доехала: за эти сутки больше не повторяем."""
+        self.summary_day = report.day
+
 
 def _iso(at: datetime | None) -> str | None:
     return at.isoformat() if at else None
@@ -381,6 +519,7 @@ def read_stack(stack: str, log: dict[str, Any]) -> StackScan:
             sample=(before.sample if before else redact(rest)[:SAMPLE_CHARS]),
             at=at or (before.at if before else None),
             count=(before.count if before else 0) + 1,
+            times=((*before.times, at) if before else (at,)),
         )
     return StackScan(
         stack=stack,
@@ -404,14 +543,41 @@ def _forget(seen: Seen, now: datetime) -> None:
             del seen.entries[key]
 
 
-def digest(seen: Seen, sweep: Sweep) -> Report:
-    """Сравнить проход с прошлым: что новое, что стихло, о чём молчать."""
+def _roll(seen: Seen, day: str) -> None:
+    """Сутки сменились: закрытый день уезжает в `day_before`, счёт с нуля."""
+    if seen.day == day:
+        return
+    if seen.day is not None:
+        for entry in seen.entries.values():
+            entry.day_earlier = entry.day_before
+            entry.day_before = entry.day_count
+            entry.day_count = 0
+    seen.day = day
+
+
+def _surging(entry: Entry, gain: int, at: datetime) -> bool:
+    """Стал ли знакомый фон снова новостью: в разы выше собственной нормы."""
+    if not entry.told or entry.quieted is not None:
+        return False
+    if entry.passes < SURGE_PASSES or gain < SURGE_MIN:
+        return False
+    if gain < SURGE_FACTOR * entry.norm:
+        return False
+    return entry.surged is None or at - entry.surged > SURGE_HOLD
+
+
+def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
+    """Сравнить проход с прошлым: что новое, что всплеск, что стихло, что длится."""
     gaps = sweep.gaps(seen.last_pass)
     _forget(seen, sweep.at)
+    day = sweep.at.astimezone(tz).date().isoformat()
+    _roll(seen, day)
     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
+    born: set[str] = set()
+    surge: list[Surge] = []
     for key, hit in hits.items():
         entry = seen.entries.get(key)
         if entry is None:
@@ -421,24 +587,54 @@ def digest(seen: Seen, sweep: Sweep) -> Report:
                 service=hit.service,
                 signal=hit.signal,
                 sample=hit.sample,
-                first=hit.at or sweep.at,
+                first=hit.times[0] if hit.times else sweep.at,
                 looked=sweep.at,
             )
             seen.entries[key] = entry
+            born.add(key)
             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
+        gain = hit.since(entry.last)
+        if not gain:
+            # Та же строка, что и в прошлый проход: хвост перечитан, беда не росла.
+            continue
+        if _surging(entry, gain, sweep.at):
+            surge.append(Surge(entry=entry, gain=gain, norm=entry.norm))
+        if entry.quieted is not None:
+            # Стихшая беда вернулась: это снова новость, а не продолжение.
+            entry.quieted = None
+            entry.told = False
+        entry.count += gain
+        entry.passes += 1
+        entry.day_count += gain
+        entry.last = hit.at or entry.last or sweep.at
     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
+        if e.announced
+        and e.quieted is None
+        and e.stack in looked
+        and e.key not in born
+        and (
+            e.key not in hits
+            or (e.last is not None and sweep.at - e.last > QUIET_AFTER)
+        )
+    )
+    # Длящееся: молчит в проходе, но живёт в состоянии и едет в суточную сводку.
+    ongoing = tuple(
+        e
+        for e in seen.entries.values()
+        if e.quieted is None
+        and e.day_before > 0
+        and e.last is not None
+        and sweep.at - e.last <= QUIET_AFTER
     )
     seen.last_pass = sweep.at
     # Холодный старт без единого прочитанного стека базой не считается: иначе
@@ -450,7 +646,11 @@ def digest(seen: Seen, sweep: Sweep) -> Report:
     return Report(
         sweep=sweep,
         new=tuple(sorted(fresh, key=lambda e: (e.stack, e.service))),
+        surge=tuple(sorted(surge, key=lambda s: -s.gain)),
         quiet=tuple(sorted(quiet, key=lambda e: (e.stack, e.service))),
+        ongoing=tuple(sorted(ongoing, key=lambda e: -e.day_before)),
+        summary=bool(ongoing) and seen.summary_day != day,
+        day=day,
         trouble=trouble,
         gaps=gaps,
         cold=cold,
@@ -505,6 +705,66 @@ def _entry_line(entry: Entry, tz: tzinfo, *, new: bool) -> str:
     return head + entry.sample
 
 
+def _surge_line(item: Surge, tz: tzinfo) -> str:
+    entry = item.entry
+    return (
+        f"- {entry.where} {_when(entry.last, tz)} {entry.signal}: "
+        f"×{item.gain} за проход при норме ×{item.norm:.1f}, "
+        f"идёт с {_when(entry.first, tz)} - {entry.sample}"
+    )
+
+
+def grouped(entries: tuple[Entry, ...]) -> tuple[Ongoing, ...]:
+    """Слить длящееся по стеку, сервису и сигналу - рядов, а не строк."""
+    rows: dict[tuple[str, str], list[Entry]] = {}
+    for entry in entries:
+        rows.setdefault((entry.where, entry.signal), []).append(entry)
+    out = [
+        Ongoing(
+            where=where,
+            signal=signal,
+            first=min((e.first for e in group if e.first), default=None),
+            day_before=sum(e.day_before for e in group),
+            day_earlier=sum(e.day_earlier for e in group),
+            kinds=len(group),
+        )
+        for (where, signal), group in rows.items()
+    ]
+    return tuple(sorted(out, key=lambda o: -o.day_before))
+
+
+def _daily_line(row: Ongoing, tz: tzinfo) -> str:
+    kinds = f", строк {row.kinds} видов" if row.kinds > 1 else ""
+    return (
+        f"- {row.where} {row.signal}{kinds}, с {_when(row.first, tz)}, "
+        f"за сутки {row.day_before}, {row.trend()}"
+    )
+
+
+def daily(report: Report, tz: tzinfo = UTC) -> str:
+    """Суточная сводка по длящемуся: имена и числа, без строк лога.
+
+    Отдельный инжект и отдельная мысль: в проходе это молчит, но раз в сутки
+    мастер видит, что именно идёт до сих пор и в какую сторону.
+    """
+    rows = grouped(report.ongoing)
+    lines = [
+        f"{SELF_MARK}: длящееся, сводка за {report.day}.",
+        (
+            f"Идёт до сих пор: {len(rows)} рядов "
+            f"({len(report.ongoing)} разных строк), за закрытые сутки:"
+        ),
+    ]
+    lines.extend(_daily_line(row, tz) for row in rows[:MAX_DAILY])
+    if len(rows) > MAX_DAILY:
+        lines.append(f"…и ещё {len(rows) - MAX_DAILY} рядов.")
+    lines.append(
+        "Новым это не считается и в проходах молчит; строки смотри "
+        "komodo(logs) или komodo(search_logs)."
+    )
+    return "\n".join(lines)
+
+
 def brief(report: Report, tz: tzinfo = UTC) -> str:
     """Инжект: что нового, где, сколько раз - и всегда область просмотра."""
     sweep = report.sweep
@@ -531,8 +791,14 @@ def brief(report: Report, tz: tzinfo = UTC) -> str:
         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:
+    elif not report.cold and sweep.read and not report.surge:
         lines.append("Нового в этой области нет.")
+    if report.surge:
+        lines.append(
+            f"Всплеск знакомого ({len(report.surge)}) - строка не новая, "
+            "но её стало в разы больше своей нормы:"
+        )
+        lines.extend(_surge_line(item, tz) for item in report.surge[:MAX_NEW])
     if report.quiet:
         lines.append(f"Стихло ({len(report.quiet)}), время - события, не прохода:")
         lines.extend(_entry_line(e, tz, new=False) for e in report.quiet[:MAX_QUIET])