beaver-agent/tests/test_komodo_logs.py

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,
    SURGE_MIN,
    Seen,
    StackScan,
    Sweep,
    brief,
    daily,
    digest,
    grouped,
    note,
    read_stack,
    record,
)
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 "хвост 2000 строк на стек: строки с 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_named_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")),
    )
    (hit,) = scan.hits
    assert "ghp_ZZ123" not in hit.sample
    assert hit.sample.startswith("ERROR auth failed token=")


def test_a_bare_blob_stays_readable():
    # Размен с открытыми глазами: id задач, сессий и коммитов - это сигнал,
    # и замазывать их ради редкого безымянного ключа дороже, чем оставить.
    scan = read_stack(
        "komodo",
        log(line("svc", "2026-09-20T17:40:00.0", "ERROR task kJ8sdfLKJ23kjhsdf98234x")),
    )
    (hit,) = scan.hits
    assert "kJ8sdfLKJ23kjhsdf98234x" in hit.sample


def stamps(day, hour, count, start=0):
    return [
        f"2026-09-{day:02d}T{hour:02d}:{m:02d}:00.0"
        for m in range(start, start + count)
    ]


def burst(stack, times, text="ERROR всё та же беда", service="svc"):
    """Хвост стека, где одна и та же беда повторилась в перечисленные минуты."""
    return read_stack(stack, log(*(line(service, t, text) for t in times)))


def at(day, hour, minute=0):
    return datetime(2026, 9, day, hour, minute, tzinfo=UTC)


def told(state, report):
    state.mark_told(report)
    return report


def test_a_continuing_trouble_is_told_once_and_then_keeps_quiet(tmp_path):
    state = seen(tmp_path)
    first = told(
        state, digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, 10)))
    )
    assert len(first.new) == 1

    # Хвост перечитан: те же строки плюс свежие - беда идёт, но новостью не стала.
    later = digest(
        state,
        swept([burst("gitea", stamps(20, 9, 3) + stamps(20, 12, 4))], at=at(20, 13)),
    )
    assert later.new == () and later.surge == () and later.quiet == ()
    assert not later.worth_telling
    assert later.ongoing == ()  # за прошлые сутки её ещё не было


def test_the_reread_tail_does_not_inflate_the_count(tmp_path):
    state = seen(tmp_path)
    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, 10))))
    for hour in (13, 16, 19):
        digest(state, swept([burst("gitea", stamps(20, 9, 3))], at=at(20, hour)))
    (entry,) = state.entries.values()
    assert entry.count == 3  # три повтора, а не три на каждый из четырёх проходов
    assert entry.passes == 1


def test_a_continuing_trouble_comes_back_in_the_daily_summary(tmp_path):
    state = seen(tmp_path)
    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 5))], at=at(20, 10))))

    # Первый проход следующих суток - 01:40, он же несёт сводку.
    night = digest(
        state,
        swept([burst("gitea", stamps(20, 9, 5) + stamps(21, 1, 2))], at=at(21, 1, 40)),
    )
    assert night.summary and len(night.ongoing) == 1
    assert night.new == () and not night.worth_telling
    text = daily(night, UTC)
    assert "сводка за 2026-09-20" in text  # числа - про закрывшиеся сутки
    assert "gitea/svc ошибка, с 20.09 09:00, за сутки 5, первые сутки" in text
    assert "ERROR" not in text  # образцов строк в сводке нет намеренно
    told(state, night)
    state.mark_daily(night)

    # Тот же день, следующий проход: сводку уже отдали.
    again = digest(
        state,
        swept([burst("gitea", stamps(20, 9, 5) + stamps(21, 1, 2))], at=at(21, 4, 40)),
    )
    assert again.ongoing and not again.summary


def test_the_summary_says_which_way_it_is_going(tmp_path):
    state = seen(tmp_path)
    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 2))], at=at(20, 10))))
    day_two = swept(
        [burst("gitea", stamps(20, 9, 2) + stamps(21, 9, 8))], at=at(21, 10)
    )
    told(state, digest(state, day_two))
    third = digest(
        state,
        swept(
            [burst("gitea", stamps(20, 9, 2) + stamps(21, 9, 8) + stamps(22, 1, 1))],
            at=at(22, 1, 40),
        ),
    )
    assert "за сутки 8, растёт (было 2)" in daily(third, UTC)


def test_a_surge_of_the_familiar_breaks_the_silence(tmp_path):
    state = seen(tmp_path)
    state_report = digest(
        state, swept([burst("gitea", stamps(20, 9, 1))], at=at(20, 10))
    )
    told(state, state_report)
    for hour, minute in ((13, 12), (16, 15)):
        digest(
            state,
            swept([burst("gitea", stamps(20, hour - 1, 1, minute))], at=at(20, hour)),
        )
    (entry,) = state.entries.values()
    assert entry.passes == 3 and entry.norm == 1.0

    spike = digest(state, swept([burst("gitea", stamps(20, 18, 12))], at=at(20, 19)))
    (item,) = spike.surge
    assert item.gain == 12 and item.norm == 1.0
    assert spike.new == () and spike.worth_telling
    text = brief(spike)
    assert "Всплеск знакомого (1)" in text
    assert "×12 за проход при норме ×1.0" in text
    assert "Нового в этой области нет" not in text

    # Второй такой же проход внутри выдержки молчит: это уже не новость.
    told(state, spike)
    more = digest(state, swept([burst("gitea", stamps(20, 21, 12))], at=at(20, 22)))
    assert more.surge == () and not more.worth_telling


def test_a_small_jump_is_not_a_surge(tmp_path):
    state = seen(tmp_path)
    told(state, digest(state, swept([burst("gitea", stamps(20, 9, 1))], at=at(20, 10))))
    for hour in (13, 16):
        digest(
            state,
            swept([burst("gitea", stamps(20, hour - 1, 1, hour))], at=at(20, hour)),
        )
    # Вчетверо больше, но всего четыре штуки: пол в SURGE_MIN держит тишину.
    small = digest(state, swept([burst("gitea", stamps(20, 18, 4))], at=at(20, 19)))
    assert small.surge == () and not small.worth_telling


def test_a_trouble_gone_silent_is_reported_even_while_its_line_lingers(tmp_path):
    state = seen(tmp_path)
    tail = burst("gitea", stamps(20, 9, 2))
    told(state, digest(state, swept([tail], at=at(20, 10))))

    # Сутки та же строка лежит в хвосте, новых событий нет.
    assert not digest(state, swept([tail], at=at(20, 22))).quiet
    stopped = digest(state, swept([tail], at=at(21, 12)))
    (gone,) = stopped.quiet
    assert gone.last == at(20, 9, 1)
    assert "последнее в 20.09 09:01" in brief(stopped)

    told(state, stopped)
    assert digest(state, swept([tail], at=at(21, 16))).quiet == ()


def test_a_trouble_that_returns_after_the_all_clear_is_new_again(tmp_path):
    state = seen(tmp_path)
    tail = burst("gitea", stamps(20, 9, 2))
    told(state, digest(state, swept([tail], at=at(20, 10))))
    told(state, digest(state, swept([tail], at=at(21, 12))))  # стихла

    back = digest(
        state,
        swept([burst("gitea", stamps(20, 9, 2) + stamps(21, 18, 3))], at=at(21, 19)),
    )
    assert len(back.new) == 1 and back.worth_telling


def test_the_background_of_the_baseline_never_claims_to_have_stopped(tmp_path):
    # Про фон, принятый в базовую линию, мастер не слышал - «стихло» про него врёт.
    state = seen(tmp_path, known=())
    state.fresh = False
    cold = digest(state, swept([burst("lodge", stamps(20, 9, 2))], at=at(20, 10)))
    assert cold.new == () and cold.baseline == 1
    gone = digest(state, swept([read_stack("lodge", log())], at=at(20, 13)))
    assert gone.quiet == () and not gone.worth_telling


def test_the_summary_folds_the_variants_of_one_service_into_one_row(tmp_path):
    state = seen(tmp_path)
    # Разные по смыслу строки: отличие в одних цифрах - это одна и та же беда.
    variants = [
        burst("gitea", stamps(20, 9, 4), text=f"ERROR {what}")
        for what in ("push отклонён", "миграция не прошла", "ssh-ключ отвергнут")
    ]
    told(state, digest(state, swept(variants, at=at(20, 10))))
    night = digest(state, swept(variants, at=at(21, 1, 40)))
    (row,) = grouped(night.ongoing)
    assert row.kinds == 3 and row.day_before == 12
    text = daily(night, UTC)
    assert "Идёт до сих пор: 1 рядов (3 разных строк)" in text
    assert "gitea/svc ошибка, строк 3 видов, с 20.09 09:00, за сутки 12" in text


def test_the_daily_summary_arrives_even_when_all_is_quiet(tmp_path):
    # Мёртвый джоб и чистый флот должны выглядеть по-разному.
    state = seen(tmp_path)
    for hour in (10, 13, 16, 19, 22):
        digest(state, swept([read_stack("gitea", log())], at=at(20, hour)))
    night = digest(state, swept([read_stack("gitea", log())], at=at(21, 1, 40)))
    assert night.summary and night.ongoing == ()
    assert not night.worth_telling  # обычного инжекта нет, суточный - есть
    text = daily(night, UTC)
    assert text.splitlines() == [
        "Логи флота: сводка за 2026-09-20.",
        "Проходов за сутки 5 из 8 ожидавшихся; в последнем прочитано 1 стеков "
        "из 1 в области (dell-home, personal-main).",
        "Длящегося нет: всё, о чём я говорил, стихло или не повторялось.",
    ]


def test_the_daily_summary_counts_the_refusals(tmp_path):
    state = seen(tmp_path)
    digest(state, swept([StackScan(stack="gitea", error="timeout")], at=at(20, 10)))
    digest(state, Sweep(at=at(20, 13), unreachable="OSError: нет связи"))
    night = digest(state, swept([read_stack("gitea", log())], at=at(21, 1, 40)))
    assert (
        "Отказов доступа за те сутки: чтений стеков 1, проходов без ответа Komodo 1"
        in daily(night, UTC)
    )


def test_a_slowly_growing_trouble_still_breaks_through(tmp_path):
    # Варёная лягушка: норма замерла на молодости беды и за ростом не тянется.
    state = seen(tmp_path)
    told(state, digest(state, swept([burst("gitea", stamps(20, 8, 1))], at=at(20, 9))))
    counts = [1, 1, 2, 3, 4, 6, 9, 13]
    hour = 10
    seen_times = stamps(20, 8, 1)
    surged = []
    for n in counts:
        seen_times = seen_times + stamps(20, hour - 1, n, start=10)
        report = digest(state, swept([burst("gitea", seen_times)], at=at(20, hour)))
        if report.surge:
            surged.append((hour, report.surge[0].gain, report.surge[0].norm))
            told(state, report)
        hour += 1
    (entry,) = state.entries.values()
    assert entry.norm == 1.0  # заморожена по первым трём активным проходам
    assert surged and surged[0][1] >= SURGE_MIN
    assert surged[0][2] == 1.0


def test_the_ceiling_fires_without_any_relative_norm(tmp_path):
    # Беда, чья собственная норма уже огромна: относительный порог её не возьмёт.
    state = seen(tmp_path)
    times = stamps(20, 8, 40)
    told(state, digest(state, swept([burst("gitea", times)], at=at(20, 9))))
    for hour in (12, 15):
        times = times + stamps(20, hour - 1, 40, start=10)
        digest(state, swept([burst("gitea", times)], at=at(20, hour)))
    (entry,) = state.entries.values()
    assert entry.norm == 40.0
    huge = stamps(20, 17, 55, start=0)
    spike = digest(state, swept([burst("gitea", times + huge)], at=at(20, 18)))
    (item,) = spike.surge
    assert item.gain == 55  # 55 < 4 × 40, взял потолок, а не кратность
    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