"""Логи флота периодами: что в хвостах похоже на беду и чего там раньше не было.

Ходит и читает джоб `jobs/logs.py`, судит этот модуль. Правило одно: мастера
дёргает только НОВОЕ. То, что молотит в логе третьи сутки, уже сказано один
раз и больше не показывается; показывается то, чего в прошлом проходе не было,
и обратный переход «было → нет» - с временем последнего события, не с временем
прохода (логи берутся с `--timestamps`, время в строке настоящее).

Молчать - не то же самое, что забыть. Беда, идущая неделю, после первого
доклада не исчезает: она переходит в длящиеся и раз в сутки идёт отдельной
короткой сводкой (имена и числа, без строк), а если за проход её стало в разы
больше собственной нормы или просто больше потолка - снова дёргает сразу.
Сводка уходит и в лучшем случае, пустой: молчание что-то значит, только если
известно, что джоб жив и сколько проходов он сделал. Затихание наоборот: не тревога,
а строка «стихло» с временем последнего события.

Показанную строку лога чистит общий `security.redact` - тот же, что стоит на
логах и на выдаче MCP: правила в одном месте, и он знает литеральные значения
секретов из env. Голые блобы без имени рядом он не трогает намеренно - id
задач, сессий и коммитов это ровно то, по чему беду потом и опознают.

Показание всегда несёт свою область: сколько стеков в области, сколько из них
прочиталось, какой отрезок логов попал в хвост. Без этого «всё чисто»
неотличимо от «половину не посмотрел», поэтому чего не смог - тоже в инжекте.
"""

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_gateway.security.redact import redact

from beaver_agent.hands.komodo import ANSI

if TYPE_CHECKING:
    from pathlib import Path

__all__ = [
    "SCOPE",
    "SIGNALS",
    "TAIL",
    "Entry",
    "Ongoing",
    "Report",
    "Seen",
    "StackScan",
    "Surge",
    "Sweep",
    "brief",
    "daily",
    "digest",
    "grouped",
    "read_stack",
]

SCOPE = ("dell-home", "personal-main")
"""Свои серверы. Машины заказчиков (md-cars-*, veris-*, ms-agents-*) не наши:
там чужой прод и чужие секреты, и читать их логи в свой контекст - нельзя."""

TAIL = 2000
"""Строк на контейнер, и число из замера, а не «чтобы хватало».

Замер 20.09.2026 по всем 36 стекам области: при 500 строках хвост покрывал
меньше трёх часов (промежуток между проходами) ровно у одного стека -
`dell-home-music`, ~1170 строк/ч, то есть 51 минута; следующий по болтливости
`personal-main-caddy` покрывал 11,6 часа, `beaver-agent` - 20 ч, медиана по
области около 110 часов. 2000 строк закрывают и music (~3,4 часа) почти
бесплатно: тихие стеки отдают столько, сколько у них есть, растут три-четыре
болтливых. Где хвост всё равно короче промежутка, инжект говорит об этом сам."""

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+")
"""Числа и хеши в строке - шум: без них «то же самое» узнаётся между проходами."""

SAMPLE_CHARS = 180
MAX_NEW = 25
MAX_QUIET = 10
MAX_TROUBLE = 10
GAP_NAMES = 3
UNPARSED_LOUD = 10
"""Столько неразобранных строк на проход - уже не обрезанная голова, а дыра."""
MAX_DAILY = 20
KEEP = timedelta(days=14)
"""Сколько помнить беду, которой больше не видно: дальше пусть будет новой."""

QUIET_AFTER = timedelta(hours=24)
"""Сутки без единого нового события - беда считается стихшей, даже если её
строка ещё лежит в хвосте. Иначе на тихом стеке «стихло» ждало бы неделями,
пока строку не вытеснит."""

PASSES_A_DAY = 8
"""Сколько проходов должно случиться за сутки - ровно столько часов в кроне
(`40 1,4,7,10,13,16,19,22`). Сводка называет это число рядом с фактическим:
проходов меньше ожидаемого - значит, молчание в остальные часы ничего не
значило."""

SURGE_FACTOR = 4.0
"""Во сколько раз проход должен перебить норму беды, чтобы снова стоить
тревоги. Проходов восемь в сутки, и счёт за проход сам по себе гуляет вдвое
(деплой, ночной крон, перезапуск) - вчетверо это уже не дрожание. Хвост
ограничен, поэтому настоящий взрыв упирается в TAIL и видно скорее «стало на
порядок», чем точную кратность. Цифра из рассуждения, не из замера:
пересмотреть после первой недели живой работы."""
SURGE_MIN = 10
"""Пол в штуках: без него «было одно, стало четыре» будило бы каждый раз."""
SURGE_CEILING = 50
"""Столько повторов одной строки за проход - новость сами по себе, без всякой
относительной нормы: медленно растущий отказ норму подтягивает за собой, и
кратность у него не срабатывает никогда.

Отсчитано от живых чисел флота (20.09.2026): самый громкий ряд -
`personal-main-caddy` со 120 повторами в сутки на десять видов строк, это
~15 за проход на ряд; самая громкая отдельная строка - `coturn` с 38 в сутки,
то есть ~5 за проход. 50 - вдесятеро выше самой громкой строки и втрое выше
самого громкого ряда: известный фон под него не попадает, а настоящий взрыв
(~400 в сутки одной строкой) попадает. Тоже из рассуждения, а не из замера
взрыва - пересмотреть после первой недели."""
NORM_PASSES = 3
"""По скольким первым активным проходам берётся норма беды - и на этом она
замирает. Пересчитывать её дальше нельзя: отказ, растущий постепенно, тянет
свою норму за собой, и порог не срабатывает никогда - варёная лягушка."""
SURGE_HOLD = timedelta(hours=12)
"""Одна беда - не чаще раза в полсуток, даже если растёт ступенями."""
TREND_UP, TREND_DOWN = 1.5, 0.67
"""Полтора раза в сутках - уже тренд, между ними - то же самое."""

KEY_CHARS = 12
STATE_VERSION = 3
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
    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)
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
    """Всего повторов, посчитанных приростом, а не перечитыванием хвоста."""
    passes: int = 0
    """В скольких проходах она прирастала."""
    norm: float = 0.0
    """Норма: среднее за первые ``NORM_PASSES`` активных проходов, дальше
    заморожена. Считается по молодости беды, пока она ещё не разрослась."""
    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}"


@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:
    """Что из прохода стоит мастера."""

    sweep: Sweep
    new: tuple[Entry, ...] = ()
    surge: tuple[Surge, ...] = ()
    quiet: tuple[Entry, ...] = ()
    ongoing: tuple[Entry, ...] = ()
    """Длящееся: молчит в обычном проходе, идёт в суточную сводку."""
    summary: bool = False
    """Сутки закрылись и сводку за них ещё не отдавали. Она уходит всегда -
    и когда длящегося нет: молчание должно быть отличимо от мёртвого джоба."""
    day: str = ""
    """Закрывшиеся сутки: это про них все числа сводки."""
    today: str = ""
    """Текущие сутки: ими помечается, что сводку уже отдали."""
    passes: int = 0
    """Проходов за закрывшиеся сутки - против ``PASSES_A_DAY`` ожидавшихся."""
    misses: int = 0
    """Чтений стеков, не удавшихся за те сутки."""
    blind: int = 0
    """Проходов тех суток, в которых Komodo не ответил вовсе."""
    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.surge 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
    day: str | None = None
    """Локальные сутки, которым принадлежат суточные счётчики записей."""
    closed: str | None = None
    """Последние закрывшиеся сутки: за них и отчитывается сводка."""
    passes_today: int = 0
    passes_before: int = 0
    """Сколько проходов случилось за текущие и за закрытые сутки: сводка
    называет это число - иначе отвалившийся крон выглядит как чистый флот."""
    misses_today: int = 0
    misses_before: int = 0
    """Сколько чтений стеков не удалось за сутки."""
    blind_today: int = 0
    blind_before: int = 0
    """Сколько проходов за сутки Komodo не ответил вовсе."""
    summary_day: str | 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)),
                passes=int(row.get("passes", 0)),
                norm=float(row.get("norm", 0.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()
        }
        return cls(
            path=path,
            entries=entries,
            baselined=set(raw.get("baselined", ())),
            trouble=tuple(raw.get("trouble", ())),
            last_pass=_parse(raw.get("last_pass")),
            day=raw.get("day"),
            closed=raw.get("closed"),
            passes_today=int(raw.get("passes_today", 0)),
            passes_before=int(raw.get("passes_before", 0)),
            misses_today=int(raw.get("misses_today", 0)),
            misses_before=int(raw.get("misses_before", 0)),
            blind_today=int(raw.get("blind_today", 0)),
            blind_before=int(raw.get("blind_before", 0)),
            summary_day=raw.get("summary_day"),
            fresh=False,
        )

    def save(self) -> None:
        body = {
            "version": STATE_VERSION,
            "last_pass": _iso(self.last_pass),
            "day": self.day,
            "closed": self.closed,
            "passes_today": self.passes_today,
            "passes_before": self.passes_before,
            "misses_today": self.misses_today,
            "misses_before": self.misses_before,
            "blind_today": self.blind_today,
            "blind_before": self.blind_before,
            "summary_day": self.summary_day,
            "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,
                    "passes": e.passes,
                    "norm": e.norm,
                    "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()
            },
        }
        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
            entry.announced = True
        for item in report.surge:
            item.entry.surged = report.sweep.at
        for entry in report.quiet:
            entry.quieted = report.sweep.at
        self.trouble = report.trouble

    def mark_daily(self, report: Report) -> None:
        """Сводка доехала: в эти сутки больше не повторяем."""
        self.summary_day = report.today


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 _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 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,
        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 _roll(seen: Seen, day: str) -> None:
    """Сутки сменились: закрытый день уезжает в `*_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.closed = seen.day
        seen.passes_before, seen.passes_today = seen.passes_today, 0
        seen.misses_before, seen.misses_today = seen.misses_today, 0
        seen.blind_before, seen.blind_today = seen.blind_today, 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.surged is not None and at - entry.surged <= SURGE_HOLD:
        return False
    if gain >= SURGE_CEILING:
        return True
    return (
        entry.passes >= NORM_PASSES
        and gain >= SURGE_MIN
        and gain >= SURGE_FACTOR * entry.norm
    )


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)
    seen.passes_today += 1
    seen.misses_today += max(sweep.in_scope - len(sweep.read), 0)
    seen.blind_today += 1 if sweep.unreachable else 0
    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:
            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
            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
    # Стихло - только про то, о чём говорили, и только у прочитанных стеков:
    # чего не видели, того не судим, а про фон мастер и не слышал.
    quiet = tuple(
        e
        for e in seen.entries.values()
        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
    # Холодный старт без единого прочитанного стека базой не считается: иначе
    # первый удачный проход вывалит всю область как «новое».
    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))),
        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=seen.closed is not None and seen.summary_day != day,
        day=seen.closed or "",
        today=day,
        passes=seen.passes_before,
        misses=seen.misses_before,
        blind=seen.blind_before,
        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 _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:
    """Суточная сводка: имена и числа, без строк лога.

    Уходит раз в сутки всегда, даже когда длящегося нет и всё чисто: строка
    про проходы и охват - единственное, чем здоровая тишина отличается от
    отвалившегося крона, побитого состояния и не поднявшегося стека.
    """
    sweep = report.sweep
    rows = grouped(report.ongoing)
    lines = [
        f"{SELF_MARK}: сводка за {report.day}.",
        (
            f"Проходов за сутки {report.passes} из {PASSES_A_DAY} ожидавшихся; "
            f"в последнем прочитано {len(sweep.read)} стеков из {sweep.in_scope} "
            f"в области ({', '.join(sweep.scope)})."
        ),
    ]
    if report.misses or report.blind:
        lines.append(
            f"Отказов доступа за те сутки: чтений стеков {report.misses}, "
            f"проходов без ответа Komodo {report.blind}."
        )
    if not rows:
        lines.append("Длящегося нет: всё, о чём я говорил, стихло или не повторялось.")
        return "\n".join(lines)
    lines.append(
        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
    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 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])
        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)