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

beaver_agent/hands/komodo_logs.py

@@ -55,7 +55,9 @@ __all__ = [
     "daily",
     "digest",
     "grouped",
+    "note",
     "read_stack",
+    "record",
 ]
 
 SCOPE = ("dell-home", "personal-main")
@@ -148,6 +150,30 @@ NORM_PASSES = 3
 свою норму за собой, и порог не срабатывает никогда - варёная лягушка."""
 SURGE_HOLD = timedelta(hours=12)
 """Одна беда - не чаще раза в полсуток, даже если растёт ступенями."""
+
+MUTE_KINDS = 15
+"""Столько новых ВИДОВ строк от одного стека за проход - это не пятнадцать
+бед, а беда, печатающая каждый раз неповторимый текст (нормализация чисел и
+хешей такую не схлопывает). Дальше стек глушится - но громко, строкой в
+инжекте, а не молча.
+
+Отсчитано от живых чисел (20.09.2026): самый шумный на новые виды стек -
+`komodo`, 67 видов за сутки, то есть ~8 за проход; `personal-main-caddy` -
+12 за сутки; у остальных 0-1. 15 за проход - вдвое выше худшего известного
+фона. Цифра из рассуждения, не из замера шторма: пересмотреть после недели."""
+MUTE_SHOW = 3
+"""Сколько видов из шторма всё-таки показать - чтобы было видно, что глушится."""
+MUTE_CLEAR = 3
+"""Поток спал до стольких видов за проход - глушение снимается."""
+MUTE_MIN = timedelta(hours=24)
+"""Но не раньше суток: шторм идёт рывками, и без выдержки глушение мигало бы
+через проход, а каждое включение - строка в инжекте."""
+
+JOURNAL_MAX = 2_000_000
+"""Байт на файл журнала, дальше - поворот в `.1`. При восьми проходах в сутки
+это месяцы истории в двух файлах: пороги пересматривать будет по чему."""
+JOURNAL_TROUBLES = 50
+"""Сколько активных бед писать в строку прогона - самые громкие."""
 TREND_UP, TREND_DOWN = 1.5, 0.67
 """Полтора раза в сутках - уже тренд, между ними - то же самое."""
 
@@ -288,6 +314,38 @@ class Entry:
         return f"{self.stack}/{self.service}"
 
 
+@dataclass(frozen=True, slots=True)
+class Active:
+    """Беда, которая прирастала в этом проходе: строка журнала прогонов."""
+
+    entry: Entry
+    gain: int
+
+
+@dataclass(slots=True)
+class Mute:
+    """Приглушённый стек: сколько видов проглочено и с каких пор."""
+
+    stack: str
+    since: datetime
+    kinds: int = 0
+    """Всего видов, проглоченных с начала глушения."""
+    day_kinds: int = 0
+    day_before: int = 0
+    told: bool = False
+    """Сказали ли мастеру, что глушим. Не сказали - повторим в следующий проход:
+    молчаливое глушение - это ровно тот отказ, против которого оно и сделано."""
+
+
+@dataclass(frozen=True, slots=True)
+class Storm:
+    """Шторм новых видов от одного стека - одной строкой вместо сорока."""
+
+    stack: str
+    kinds: int
+    shown: int
+
+
 @dataclass(frozen=True, slots=True)
 class Ongoing:
     """Длящееся одним рядом: стек, сервис и сигнал, а не отдельная строка.
@@ -334,6 +392,14 @@ class Report:
     quiet: tuple[Entry, ...] = ()
     ongoing: tuple[Entry, ...] = ()
     """Длящееся: молчит в обычном проходе, идёт в суточную сводку."""
+    active: tuple[Active, ...] = ()
+    """Кто прирастал в этом проходе - это и пишется в журнал прогонов."""
+    storms: tuple[Storm, ...] = ()
+    """Стеки, чей поток новых видов пришлось заглушить в этом проходе."""
+    unmuted: tuple[str, ...] = ()
+    """Стеки, у которых поток спал и глушение снято."""
+    muted: tuple[Mute, ...] = ()
+    """Кто приглушён сейчас - напоминание раз в сутки, пока не спадёт."""
     summary: bool = False
     """Сутки закрылись и сводку за них ещё не отдавали. Она уходит всегда -
     и когда длящегося нет: молчание должно быть отличимо от мёртвого джоба."""
@@ -359,7 +425,15 @@ class Report:
     @property
     def worth_telling(self) -> bool:
         """Суточная сводка сюда не входит: она едет своим инжектом."""
-        return bool(self.new or self.surge 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
+            or self.storms
+            or self.unmuted
+        )
 
 
 @dataclass(slots=True)
@@ -370,6 +444,8 @@ class Seen:
     entries: dict[str, Entry] = field(default_factory=dict)
     baselined: set[str] = field(default_factory=set)
     """Стеки, чьё прошлое уже принято за данность: их старое не «новое»."""
+    mutes: dict[str, Mute] = field(default_factory=dict)
+    """Стеки в приглушённом режиме: их новые виды глотаются, но считаются."""
     trouble: tuple[str, ...] = ()
     """О каких отказах уже сказали: Komodo лежит - скажем один раз, не каждые 3 ч."""
     last_pass: datetime | None = None
@@ -427,6 +503,17 @@ class Seen:
             path=path,
             entries=entries,
             baselined=set(raw.get("baselined", ())),
+            mutes={
+                stack: Mute(
+                    stack=stack,
+                    since=_parse(row.get("since")) or EPOCH,
+                    kinds=int(row.get("kinds", 0)),
+                    day_kinds=int(row.get("day_kinds", 0)),
+                    day_before=int(row.get("day_before", 0)),
+                    told=bool(row.get("told")),
+                )
+                for stack, row in raw.get("mutes", {}).items()
+            },
             trouble=tuple(raw.get("trouble", ())),
             last_pass=_parse(raw.get("last_pass")),
             day=raw.get("day"),
@@ -455,6 +542,16 @@ class Seen:
             "blind_before": self.blind_before,
             "summary_day": self.summary_day,
             "baselined": sorted(self.baselined),
+            "mutes": {
+                stack: {
+                    "since": _iso(m.since),
+                    "kinds": m.kinds,
+                    "day_kinds": m.day_kinds,
+                    "day_before": m.day_before,
+                    "told": m.told,
+                }
+                for stack, m in self.mutes.items()
+            },
             "trouble": list(self.trouble),
             "entries": {
                 key: {
@@ -497,6 +594,10 @@ class Seen:
             item.entry.surged = report.sweep.at
         for entry in report.quiet:
             entry.quieted = report.sweep.at
+        for storm in report.storms:
+            mute = self.mutes.get(storm.stack)
+            if mute is not None:
+                mute.told = True
         self.trouble = report.trouble
 
     def mark_daily(self, report: Report) -> None:
@@ -609,6 +710,8 @@ def _roll(seen: Seen, day: str) -> None:
             entry.day_earlier = entry.day_before
             entry.day_before = entry.day_count
             entry.day_count = 0
+        for mute in seen.mutes.values():
+            mute.day_before, mute.day_kinds = mute.day_kinds, 0
         seen.closed = seen.day
         seen.passes_before, seen.passes_today = seen.passes_today, 0
         seen.misses_before, seen.misses_today = seen.misses_today, 0
@@ -616,6 +719,34 @@ def _roll(seen: Seen, day: str) -> None:
     seen.day = day
 
 
+def _mutes(
+    seen: Seen, kinds: dict[str, int], at: datetime
+) -> tuple[tuple[Storm, ...], tuple[str, ...], dict[str, int]]:
+    """Кого глушим, с кого глушение снимаем и сколько видов всё же показать.
+
+    Глушение никогда не бывает тихим: пока про него не сказали, оно едет в
+    инжект каждый проход, а пока держится - напоминает о себе в суточной сводке.
+    """
+    storms: list[Storm] = []
+    unmuted: list[str] = []
+    budget: dict[str, int] = {}
+    for stack, mute in list(seen.mutes.items()):
+        calm = kinds.get(stack, 0) <= MUTE_CLEAR
+        if calm and mute.told and at - mute.since >= MUTE_MIN:
+            del seen.mutes[stack]
+            unmuted.append(stack)
+    for stack, count in kinds.items():
+        if stack in seen.mutes:
+            if not seen.mutes[stack].told:
+                storms.append(Storm(stack=stack, kinds=count, shown=MUTE_SHOW))
+            continue
+        if count > MUTE_KINDS:
+            seen.mutes[stack] = Mute(stack=stack, since=at)
+            budget[stack] = MUTE_SHOW
+            storms.append(Storm(stack=stack, kinds=count, shown=MUTE_SHOW))
+    return tuple(storms), tuple(sorted(unmuted)), budget
+
+
 def _surging(entry: Entry, gain: int, at: datetime) -> bool:
     """Стал ли знакомый фон снова новостью.
 
@@ -625,6 +756,11 @@ def _surging(entry: Entry, gain: int, at: datetime) -> bool:
     """
     if not entry.told or entry.quieted is not None:
         return False
+    if not entry.passes:
+        # Первая встреча: в хвосте лежит суточный запас строк, и «прирост» тут
+        # говорит про длину хвоста, а не про скорость беды. Ни кратность, ни
+        # потолок к нему не применимы - это база, а не всплеск.
+        return False
     if entry.surged is not None and at - entry.surged <= SURGE_HOLD:
         return False
     if gain >= SURGE_CEILING:
@@ -636,6 +772,76 @@ def _surging(entry: Entry, gain: int, at: datetime) -> bool:
     )
 
 
+@dataclass(slots=True)
+class _Take:
+    """Что проход сделал с записями состояния: для сборки отчёта."""
+
+    baseline: int = 0
+    born: set[str] = field(default_factory=set)
+    surge: list[Surge] = field(default_factory=list)
+    active: list[Active] = field(default_factory=list)
+
+
+def _absorb(
+    seen: Seen,
+    sweep: Sweep,
+    hits: dict[str, Hit],
+    newcomers: set[str],
+    budget: dict[str, int],
+) -> _Take:
+    """Разложить находки прохода по состоянию: прирост, норма, глушение."""
+    take = _Take()
+    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.times[0] if hit.times else sweep.at,
+                looked=sweep.at,
+            )
+            seen.entries[key] = entry
+            take.born.add(key)
+            mute = seen.mutes.get(hit.stack)
+            if mute is not None:
+                # Шторм: показываем первые несколько, остальное глотаем - но в
+                # счёт, и стек идёт в суточную сводку как приглушённый.
+                if budget.get(hit.stack, 0) > 0:
+                    budget[hit.stack] -= 1
+                else:
+                    entry.told = True
+                    mute.kinds += 1
+                    mute.day_kinds += 1
+            if hit.stack in newcomers:
+                # Стек читаем впервые: в хвосте лежит всё его прошлое, и вываливать
+                # его как «новое» - врать. Принимаем за данность, дальше - только рост.
+                entry.told = True
+                take.baseline += 1
+        entry.looked = sweep.at
+        gain = hit.since(entry.last)
+        if not gain:
+            # Та же строка, что и в прошлый проход: хвост перечитан, беда не росла.
+            continue
+        if _surging(entry, gain, sweep.at):
+            take.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
+        if entry.passes <= NORM_PASSES:
+            # Норма берётся по молодости беды и дальше не трогается.
+            entry.norm = entry.count / entry.passes
+        entry.last = hit.at or entry.last or sweep.at
+        take.active.append(Active(entry=entry, gain=gain))
+    return take
+
+
 def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
     """Сравнить проход с прошлым: что новое, что всплеск, что стихло, что длится."""
     gaps = sweep.gaps(seen.last_pass)
@@ -648,46 +854,12 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
     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
+    kinds: dict[str, int] = {}
-    born: set[str] = set()
-    surge: list[Surge] = []
     for key, hit in hits.items():
-        entry = seen.entries.get(key)
+        if key not in seen.entries and hit.stack not in newcomers:
-        if entry is None:
+            kinds[hit.stack] = kinds.get(hit.stack, 0) + 1
-            entry = Entry(
+    storms, unmuted, budget = _mutes(seen, kinds, sweep.at)
-                key=key,
+    take = _absorb(seen, sweep, hits, newcomers, budget)
-                stack=hit.stack,
-                service=hit.service,
-                signal=hit.signal,
-                sample=hit.sample,
-                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.looked = sweep.at
-        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
-        if entry.passes <= NORM_PASSES:
-            # Норма берётся по молодости беды и дальше не трогается.
-            entry.norm = entry.count / entry.passes
-        entry.last = hit.at or entry.last or sweep.at
     seen.baselined |= looked
     # Стихло - только про то, о чём говорили, и только у прочитанных стеков:
     # чего не видели, того не судим, а про фон мастер и не слышал.
@@ -697,7 +869,7 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
         if e.announced
         and e.quieted is None
         and e.stack in looked
-        and e.key not in born
+        and e.key not in take.born
         and (
             e.key not in hits
             or (e.last is not None and sweep.at - e.last > QUIET_AFTER)
@@ -722,9 +894,13 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> 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)),
+        surge=tuple(sorted(take.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)),
+        active=tuple(sorted(take.active, key=lambda a: -a.gain)),
+        storms=storms,
+        unmuted=unmuted,
+        muted=tuple(seen.mutes.values()),
         summary=seen.closed is not None and seen.summary_day != day,
         day=seen.closed or "",
         today=day,
@@ -734,11 +910,61 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
         trouble=trouble,
         gaps=gaps,
         cold=cold,
-        baseline=baseline,
+        baseline=take.baseline,
         baselined=tuple(sorted(newcomers)),
     )
 
 
+def record(report: Report, tz: tzinfo = UTC) -> dict[str, Any]:
+    """Строка журнала прогонов: по ней через неделю пересматривают пороги."""
+    sweep = report.sweep
+    return {
+        "at": sweep.at.astimezone(tz).isoformat(timespec="seconds"),
+        "day": report.today,
+        "pass_of_day": None,
+        "stacks": len(sweep.read),
+        "in_scope": sweep.in_scope,
+        "fleet": sweep.fleet,
+        "misses": max(sweep.in_scope - len(sweep.read), 0),
+        "blind": 1 if sweep.unreachable else 0,
+        "tail": sweep.tail,
+        "lines": sum(s.lines for s in sweep.read),
+        "unparsed": sum(s.unparsed for s in sweep.read),
+        "new": len(report.new),
+        "surge": len(report.surge),
+        "quiet": len(report.quiet),
+        "ongoing": len(report.ongoing),
+        "active": len(report.active),
+        "muted": {m.stack: m.kinds for m in report.muted},
+        "troubles": [
+            {
+                "key": a.entry.key,
+                "where": a.entry.where,
+                "signal": a.entry.signal,
+                "gain": a.gain,
+                "total": a.entry.count,
+                "norm": round(a.entry.norm, 2),
+                "day": a.entry.day_count,
+            }
+            for a in report.active[:JOURNAL_TROUBLES]
+        ],
+    }
+
+
+def note(path: Path, row: dict[str, Any], *, max_bytes: int = JOURNAL_MAX) -> None:
+    """Дописать строку прогона в jsonl; файл дорос - поворот в `.1`.
+
+    Состояние - снимок и перезаписывается, по нему истории не собрать. Журнал
+    только дописывается: пороги брались из рассуждения, и пересматривать их
+    через неделю можно будет по числам, а не по памяти.
+    """
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.exists() and path.stat().st_size > max_bytes:
+        path.replace(path.with_suffix(path.suffix + ".1"))
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(row, ensure_ascii=False) + "\n")
+
+
 def _when(at: datetime | None, tz: tzinfo) -> str:
     return at.astimezone(tz).strftime("%d.%m %H:%M") if at else "время не в строке"
 
@@ -813,6 +1039,22 @@ def grouped(entries: tuple[Entry, ...]) -> tuple[Ongoing, ...]:
     return tuple(sorted(out, key=lambda o: -o.day_before))
 
 
+def _storm_line(storm: Storm) -> str:
+    return (
+        f"Стек {storm.stack}: {storm.kinds} новых ВИДОВ строк за проход - это не "
+        f"{storm.kinds} бед, а строки с неповторимым текстом. Показываю "
+        f"{storm.shown}, остальное глушу и напоминаю раз в сутки, пока не спадёт."
+    )
+
+
+def _mute_line(mute: Mute, tz: tzinfo) -> str:
+    return (
+        f"Приглушён {mute.stack}: за те сутки проглочено {mute.day_before} новых "
+        f"видов строк (всего {mute.kinds}), глушу с {_when(mute.since, tz)} - "
+        "поток пока не спал."
+    )
+
+
 def _daily_line(row: Ongoing, tz: tzinfo) -> str:
     kinds = f", строк {row.kinds} видов" if row.kinds > 1 else ""
     return (
@@ -843,6 +1085,10 @@ def daily(report: Report, tz: tzinfo = UTC) -> str:
             f"Отказов доступа за те сутки: чтений стеков {report.misses}, "
             f"проходов без ответа Komodo {report.blind}."
         )
+    lines.extend(
+        _mute_line(mute, tz)
+        for mute in sorted(report.muted, key=lambda m: -m.day_before)
+    )
     if not rows:
         lines.append("Длящегося нет: всё, о чём я говорил, стихло или не повторялось.")
         return "\n".join(lines)
@@ -894,6 +1140,11 @@ def brief(report: Report, tz: tzinfo = UTC) -> str:
             "но её стало в разы больше своей нормы:"
         )
         lines.extend(_surge_line(item, tz) for item in report.surge[:MAX_NEW])
+    lines.extend(_storm_line(storm) for storm in report.storms)
+    if report.unmuted:
+        lines.append(
+            f"Поток новых видов спал, глушение снял: {', '.join(report.unmuted)}."
+        )
     if report.quiet:
         lines.append(f"Стихло ({len(report.quiet)}), время - события, не прохода:")
         lines.extend(_entry_line(e, tz, new=False) for e in report.quiet[:MAX_QUIET])

beaver_agent/jobs/logs.py

@@ -27,7 +27,9 @@ from beaver_agent.hands.komodo_logs import (
     brief,
     daily,
     digest,
+    note,
     read_stack,
+    record,
 )
 from beaver_agent.vault import STATE_DIR, ZONE
 
@@ -39,7 +41,11 @@ if TYPE_CHECKING:
 _log = logging.getLogger(__name__)
 
 STATE_FILE = STATE_DIR / "логи-флота.json"
-"""Состояние прохода: что уже видели и о чём уже сказали."""
+"""Состояние прохода: что уже видели и о чём уже сказали. Перезаписывается."""
+JOURNAL_FILE = STATE_DIR / "логи-флота.jsonl"
+"""Журнал прогонов, только дописывается: состояние - снимок, истории в нём нет,
+а пороги всплеска и глушения взяты из рассуждения и через неделю живой работы
+пересматриваются по числам. Читать: `jq` по строкам, поля - см. `record()`."""
 PARALLEL = 4
 """Логи тянутся с periphery: четырёх параллельных хватает, больше - грубо."""
 
@@ -120,9 +126,14 @@ async def logs_watch(run: JobRun) -> None:
     ):
         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, стеков %s из %s, приглушено %s",
         len(report.new),
         len(report.surge),
         len(report.quiet),
@@ -131,4 +142,5 @@ async def logs_watch(run: JobRun) -> None:
         len(report.trouble),
         len(report.sweep.read),
         report.sweep.in_scope,
+        len(report.muted),
     )

tests/test_komodo_logs.py

@@ -1,8 +1,11 @@
+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,
@@ -14,7 +17,9 @@ from beaver_agent.hands.komodo_logs import (
     daily,
     digest,
     grouped,
+    note,
     read_stack,
+    record,
 )
 from beaver_agent.jobs.logs import sweep as sweep_fleet
 
@@ -728,3 +733,81 @@ def test_the_ceiling_fires_without_any_relative_norm(tmp_path):
     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