from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Any, cast

import pytest

from beaver_agent.hands.runs import (
    STREAK,
    Run,
    Runs,
    alarm_text,
    awake,
    clear_text,
    empty_text,
    first_line,
    look,
    mask,
    quiet_text,
)

NOW = datetime(2026, 9, 22, 15, 42, tzinfo=UTC)
BOOM = (
    "HubError: vibegram 530 bad_json: <!doctype html>\n"
    "Traceback (most recent call last):\n"
    '  File "/app/src/beaver_gateway/jobs/scheduler.py", line 170, in _dispatch\n'
    "HubError: vibegram 530 bad_json"
)


def run(rid, job, status, *, ago, error=None):
    return Run(
        id=rid,
        job=job,
        status=status,
        started_at=NOW - timedelta(minutes=ago),
        error=error if error is not None or status != "failed" else BOOM,
    )


def series(job, statuses, *, step=10, start=1000):
    """Прогоны джоба от старых к новым: statuses[-1] - последний прогон."""
    ago = step * len(statuses)
    out = []
    for i, status in enumerate(statuses):
        out.append(run(start + i, job, status, ago=ago - step * i))
    return out


def state(tmp_path, **kw):
    return Runs(path=tmp_path / "прогоны.json", **kw)


def test_three_in_a_row_is_an_alarm(tmp_path):
    rows = series("вайбграм", ["done", "failed", "failed", "failed"])
    report = look(state(tmp_path), rows, now=NOW, me="прогоны")
    assert not report.empty and report.seen == 4
    (tail,) = report.alarms
    assert tail.job == "вайбграм" and tail.count == 3
    assert tail.since == NOW - timedelta(minutes=30)
    assert not tail.capped
    text = alarm_text(report)
    assert "джоб `вайбграм` лежит 30 минут, 3 прогона подряд" in text
    assert "с 22.09 15:12 UTC" in text
    # Первая строка ошибки, а не трейсбек
    assert "HubError: vibegram 530 bad_json" in text
    assert "Traceback" not in text and "scheduler.py" not in text


def test_two_in_a_row_stays_quiet(tmp_path):
    rows = series("вайбграм", ["done", "failed", "failed"])
    report = look(state(tmp_path), rows, now=NOW, me="прогоны")
    assert not report.alarms and not report.clears and not report.empty
    # Одиночное падение - тем более молчание
    once = look(state(tmp_path), series("логи", ["failed"]), now=NOW, me="прогоны")
    assert not once.alarms


def test_only_the_tail_counts(tmp_path):
    # Падало много, но последний прогон прошёл - джоб не лежит
    rows = series("логи", ["failed", "failed", "failed", "failed", "done"])
    assert not look(state(tmp_path), rows, now=NOW, me="прогоны").alarms


def test_alarm_not_repeated_within_a_day(tmp_path):
    rows = series("вайбграм", ["done", "failed", "failed", "failed", "failed"])
    told = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=5)})
    quiet = look(told, rows, now=NOW, me="прогоны")
    assert not quiet.alarms
    assert "уже говорили меньше суток назад" in quiet.note
    older = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=25)})
    assert look(older, rows, now=NOW, me="прогоны").alarms


def test_repair_clears_and_resets(tmp_path):
    rows = series("вайбграм", ["failed", "failed", "failed", "done"])
    told = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=2)})
    report = look(told, rows, now=NOW, me="прогоны")
    assert not report.alarms
    (clear,) = report.clears
    assert clear.job == "вайбграм" and clear.count == 0 and clear.status == "done"
    assert "отбой: джоб `вайбграм` снова проходит (done)" in clear_text(report)
    # Сброс делает джоб; после него тот же вход снова может поднять тревогу
    del told.alerted["вайбграм"]
    again = look(told, series("вайбграм", ["failed"] * 3), now=NOW, me="прогоны")
    assert again.alarms


def test_repair_is_told_once_not_every_run(tmp_path):
    rows = series("вайбграм", ["failed", "failed", "failed", "done"])
    clean = state(tmp_path)
    assert not look(clean, rows, now=NOW, me="прогоны").clears


def test_flapping_job_is_a_clear_not_an_alarm(tmp_path):
    # После тревоги джоб моргает: хвост короче порога - он больше не лежит
    rows = series("вайбграм", ["done", "done", "failed", "failed"])
    told = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=2)})
    report = look(told, rows, now=NOW, me="прогоны")
    (clear,) = report.clears
    assert clear.count == 2
    assert f"порог {STREAK}" in clear_text(report)


def test_streak_running_past_the_window_says_at_least(tmp_path):
    rows = series("вайбграм", ["failed"] * 144, step=10)
    report = look(state(tmp_path), rows, now=NOW, me="прогоны")
    (tail,) = report.alarms
    assert tail.capped and tail.count == 144
    text = alarm_text(report)
    assert "лежит 24 часа и дольше, 144 прогона подряд" in text
    assert "это всё окно недели, серия началась раньше" in text


def test_empty_window_is_its_own_alarm(tmp_path):
    report = look(state(tmp_path), [], now=NOW, me="прогоны")
    assert report.empty and not report.alarms and report.seen == 0
    assert "не отработал ни один джоб" in empty_text(report)
    assert "планировщике" in empty_text(report)
    # И повторяется не чаще раза в сутки
    told = state(tmp_path, empty_alerted=NOW - timedelta(hours=5))
    assert not look(told, [], now=NOW, me="прогоны").empty
    stale = state(tmp_path, empty_alerted=NOW - timedelta(hours=25))
    assert look(stale, [], now=NOW, me="прогоны").empty


def test_own_runs_do_not_count_as_life_or_as_illness(tmp_path):
    mine = series("прогоны", ["failed", "failed", "failed", "done"])
    report = look(state(tmp_path), mine, now=NOW, me="прогоны")
    # Свои прогоны в окне есть, чужих нет: это всё равно пустое окно
    assert report.empty and not report.alarms and report.mine == 4
    assert "их 4" in empty_text(report)


def test_forgotten_job_leaves_the_state_silently(tmp_path):
    rows = series("логи", ["done", "done"])
    told = state(
        tmp_path,
        alerted={
            "исчезнувший": NOW - timedelta(days=8),
            "недавний": NOW - timedelta(hours=3),
        },
    )
    report = look(told, rows, now=NOW, me="прогоны")
    assert not report.clears  # починки мы не видели, «отбой» врать не о чем
    assert "исчезнувший" not in told.alerted
    assert "недавний" in told.alerted


def test_several_broken_jobs_come_in_one_alarm(tmp_path):
    rows = [
        *series("вайбграм", ["failed"] * 4, start=100),
        *series("логи", ["failed"] * 3, start=200),
        *series("досуг", ["done"], start=300),
    ]
    report = look(state(tmp_path), rows, now=NOW, me="прогоны")
    assert len(report.alarms) == 2
    assert report.alarms[0].job == "вайбграм"  # длиннее серия - выше в тексте
    assert alarm_text(report).count("\n") == 1


def test_first_line_masks_secrets_and_cuts_the_tail():
    assert first_line(None) == "без текста ошибки"
    assert first_line("") == "без текста ошибки"
    # Токен в строке ошибки не уезжает человеку
    assert "s3cr3tvalue" not in first_line("RuntimeError: MSOS_MCP=s3cr3tvalue died")
    assert "MSOS_MCP=<…>" in first_line("RuntimeError: MSOS_MCP=s3cr3tvalue died")
    assert "hunter2" not in first_line("ошибка: password=hunter2")
    assert "abc123" not in first_line("GET https://hub/api?t=abc123 -> 500")
    assert "https://hub/api?<…>" in first_line("GET https://hub/api?t=abc123 -> 500")
    assert "Bearer abcdefghij" not in first_line("Authorization: Bearer abcdefghij")
    # Длинная строка режется
    long = first_line("E: " + "x" * 500)
    assert len(long) == 160 and long.endswith("…")
    # Имя без запаха секрета остаётся читаемым
    assert mask("status=530 job=вайбграм") == "status=530 job=вайбграм"


def test_state_survives_a_round_trip(tmp_path):
    saved = state(
        tmp_path,
        alerted={"вайбграм": NOW, "логи": NOW - timedelta(hours=3)},
        empty_alerted=NOW,
    )
    saved.save()
    back = Runs.load(saved.path)
    assert back.alerted["вайбграм"] == NOW
    assert back.alerted["логи"] == NOW - timedelta(hours=3)
    assert back.empty_alerted == NOW
    assert Runs.load(tmp_path / "нет.json").alerted == {}
    (tmp_path / "мусор.json").write_text("не json", encoding="utf-8")
    assert Runs.load(tmp_path / "мусор.json").empty_alerted is None


class FakeRun:
    def __init__(self, database=None, *, name="прогоны", jobs=()):
        self.job = SimpleNamespace(name=name, cron="27 * * * *")
        self.conversations = SimpleNamespace(db=database)
        self.scheduler = SimpleNamespace(
            jobs=[self.job, *(SimpleNamespace(name=n, cron=c) for n, c in jobs)], tz=UTC
        )
        self.spawned = []

    async def spawn_job(self, *, agent, text, title=None):
        self.spawned.append((agent, title, text))
        return None


async def make_db(tmp_path, rows):
    from beaver_gateway.storage.db import Database
    from beaver_gateway.storage.models import JobRunRecord

    database = Database(f"sqlite+aiosqlite:///{tmp_path / 'journal.db'}")
    await database.create_all()
    async with database.session() as session:
        for row in rows:
            session.add(row)
        await session.commit()
    return database


def record(job, status, *, ago, error=None):
    from beaver_gateway.storage.models import JobRunRecord

    started = (datetime.now(UTC) - timedelta(minutes=ago)).replace(tzinfo=None)
    return JobRunRecord(
        job=job,
        trigger="cron",
        started_at=started,
        finished_at=started,
        status=status,
        error=error if error is not None or status != "failed" else BOOM,
    )


async def test_job_reads_the_real_journal_and_spawns(tmp_path, monkeypatch):
    from beaver_agent.jobs import runs as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    database = await make_db(
        tmp_path,
        [
            record("вайбграм", "failed", ago=30),
            record("вайбграм", "failed", ago=20),
            record("вайбграм", "failed", ago=10),
            record("досуг", "done", ago=5),
            record("прогоны", "done", ago=3),
            # Одно падение - не серия, сколько бы окно ни было
            record("логи", "failed", ago=60 * 30),
        ],
    )
    run_ = FakeRun(database)

    await job.watch_runs(cast("Any", run_))

    assert len(run_.spawned) == 1
    agent, title, text = run_.spawned[0]
    assert agent == "beaver-triage" and title == "прогоны"
    assert "джоб `вайбграм` лежит" in text and "3 прогона подряд" in text
    assert "логи" not in text  # одно падение
    assert "прогоны" not in text.split("\n")[1]  # сам себя не считает
    # Тревога записана - следующий прогон в те же сутки молчит
    assert Runs.load(tmp_path / "прогоны.json").alerted["вайбграм"] is not None
    await job.watch_runs(cast("Any", run_))
    assert len(run_.spawned) == 1
    await database.dispose()


async def test_job_says_otboy_when_the_job_recovers(tmp_path, monkeypatch):
    from beaver_agent.jobs import runs as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    Runs(
        path=tmp_path / "прогоны.json",
        alerted={"вайбграм": datetime.now(UTC) - timedelta(hours=2)},
    ).save()
    database = await make_db(
        tmp_path,
        [
            record("вайбграм", "failed", ago=30),
            record("вайбграм", "failed", ago=20),
            record("вайбграм", "done", ago=10),
        ],
    )
    run_ = FakeRun(database)

    await job.watch_runs(cast("Any", run_))

    assert len(run_.spawned) == 1
    _, _, text = run_.spawned[0]
    assert "отбой: джоб `вайбграм` снова проходит" in text
    assert 'urgency="normal"' in text  # отбой не будит
    assert Runs.load(tmp_path / "прогоны.json").alerted == {}
    # Сброшено - второй «отбой» не уходит
    await job.watch_runs(cast("Any", run_))
    assert len(run_.spawned) == 1
    await database.dispose()


async def test_job_alarms_on_an_empty_journal(tmp_path, monkeypatch):
    from beaver_agent.jobs import runs as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    database = await make_db(tmp_path, [record("прогоны", "done", ago=5)])
    run_ = FakeRun(database)

    await job.watch_runs(cast("Any", run_))

    assert len(run_.spawned) == 1
    _, _, text = run_.spawned[0]
    assert "нет ни одной записи" in text and "не отработал ни один джоб" in text
    assert Runs.load(tmp_path / "прогоны.json").empty_alerted is not None
    await database.dispose()


async def test_unreadable_journal_is_not_a_reading(tmp_path, monkeypatch):
    from beaver_agent.jobs import runs as job

    class Dead:
        def session(self):
            msg = "база не отвечает"
            raise RuntimeError(msg)

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    run_ = FakeRun(Dead())

    await job.watch_runs(cast("Any", run_))

    assert not run_.spawned
    assert not (tmp_path / "прогоны.json").exists()


@pytest.mark.parametrize("name", ["прогоны", "иначе-названный"])
async def test_the_watcher_never_reports_itself(tmp_path, monkeypatch, name):
    from beaver_agent.jobs import runs as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    database = await make_db(
        tmp_path,
        [
            record(name, "failed", ago=30),
            record(name, "failed", ago=20),
            record(name, "failed", ago=10),
            record("досуг", "done", ago=5),
        ],
    )
    run_ = FakeRun(database, name=name)

    await job.watch_runs(cast("Any", run_))

    assert not run_.spawned
    await database.dispose()


# --- Окно неделя: суточные джобы тоже набирают серию ---------------------------


def test_daily_job_failing_three_days_in_a_row_is_an_alarm(tmp_path):
    day = 24 * 60
    rows = [
        run(1, "закрытие", "done", ago=4 * day - 60),
        run(2, "закрытие", "failed", ago=3 * day - 60),
        run(3, "закрытие", "failed", ago=2 * day - 60),
        run(4, "закрытие", "failed", ago=day - 60),
    ]
    report = look(state(tmp_path), rows, now=NOW, me="прогоны")
    (tail,) = report.alarms
    assert tail.job == "закрытие" and tail.count == 3 and not tail.capped
    assert "джоб `закрытие` лежит 2 суток, 3 прогона подряд" in alarm_text(report)


async def test_job_reads_a_week_back(tmp_path, monkeypatch):
    from beaver_agent.jobs import runs as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    day = 24 * 60
    database = await make_db(
        tmp_path,
        [
            record("голосовой", "failed", ago=3 * day - 30),
            record("голосовой", "failed", ago=2 * day - 30),
            record("голосовой", "failed", ago=day - 30),
            # Старше недели - в окно не попадает
            record("память", "failed", ago=9 * day),
        ],
    )
    run_ = FakeRun(database)

    await job.watch_runs(cast("Any", run_))

    (spawned,) = run_.spawned
    assert "джоб `голосовой` лежит" in spawned[2]
    assert "память" not in spawned[2]
    await database.dispose()


# --- Давность: джоб перестал запускаться ---------------------------------------

CRONS = {"прогоны": "27 * * * *", "закрытие": "20 4 * * *", "вайбграм": "*/10 * * * *"}


def hourly_me(hours=72):
    """Непрерывная цепочка своих прогонов: планировщик жив всё это время."""
    return [
        run(5000 + i, "прогоны", "done", ago=60 * (hours - i) - 15)
        for i in range(hours)
    ]


def alive(ago=5):
    return run(9000, "вайбграм", "done", ago=ago)


def silent(tmp_path, *, closing_ago_h, mine=None, boot=None, **kw):
    last = run(1, "закрытие", "done", ago=closing_ago_h * 60)
    return look(
        kw.pop("runs", None) or state(tmp_path),
        [alive(), *(hourly_me() if mine is None else mine)],
        now=NOW,
        me="прогоны",
        crons=kw.pop("crons", CRONS),
        latest=kw.pop("latest", {"закрытие": last, "вайбграм": alive()}),
        tz=UTC,
        boot=boot,
    )


def test_job_silent_longer_than_two_periods_is_its_own_alarm(tmp_path):
    report = silent(tmp_path, closing_ago_h=50)
    (quiet,) = report.quiet
    assert quiet.job == "закрытие" and quiet.period == timedelta(hours=24)
    assert not report.alarms  # это не «лежит»
    text = quiet_text(report)
    assert "джоб `закрытие` не запускался 2 суток при периоде 24 часа" in text
    assert "лежит" not in text


def test_one_missed_fire_is_not_silence(tmp_path):
    # 40 часов: пропущено одно срабатывание, не два
    assert not silent(tmp_path, closing_ago_h=40).quiet


def test_silence_is_told_once_a_day_and_cleared_by_a_run(tmp_path):
    told = state(tmp_path, quiet={"закрытие": NOW - timedelta(hours=5)})
    held = silent(tmp_path, closing_ago_h=50, runs=told)
    assert not held.quiet
    assert "о молчании уже говорили меньше суток назад: закрытие" in held.note
    older = state(tmp_path, quiet={"закрытие": NOW - timedelta(hours=25)})
    assert silent(tmp_path, closing_ago_h=50, runs=older).quiet
    # Запуск появился после тревоги - «отбой», а не молчание
    back = silent(tmp_path, closing_ago_h=1, runs=told)
    (resumed,) = back.resumed
    assert resumed.job == "закрытие" and not back.clears
    assert "отбой: джоб `закрытие` снова запускается (done)" in clear_text(back)


def test_downtime_of_the_scheduler_is_not_a_missed_fire(tmp_path):
    # Гейтвей только что поднялся: после старта срабатывания ещё не было
    fresh = silent(tmp_path, closing_ago_h=50, boot=NOW - timedelta(minutes=30))
    assert not fresh.quiet
    # Сторож сам не бегал сутки (бюджет зажат, некритичное откладывается):
    # пропуски за это время - общий простой, а не молчание одного джоба
    gap = [run(1, "прогоны", "done", ago=60 * 30)]
    assert not silent(tmp_path, closing_ago_h=50, mine=gap).quiet
    assert awake(gap, now=NOW, step=timedelta(hours=2), boot=None) == NOW


def test_short_period_job_uses_the_watchers_resolution(tmp_path):
    latest = {"закрытие": run(1, "закрытие", "done", ago=60)}
    # Вайбграм раз в 10 минут, молчит 70 минут: сторож часовой, и такой
    # пропуск неотличим от его же слепоты между прогонами
    latest["вайбграм"] = run(2, "вайбграм", "done", ago=70)
    assert not silent(tmp_path, closing_ago_h=1, latest=latest).quiet
    latest["вайбграм"] = run(2, "вайбграм", "done", ago=150)
    (quiet,) = silent(tmp_path, closing_ago_h=1, latest=latest).quiet
    assert quiet.job == "вайбграм"
    assert "не запускался 2 часа при периоде 10 минут" in quiet_text(
        silent(tmp_path, closing_ago_h=1, latest=latest)
    )


def test_job_without_cron_is_not_checked(tmp_path):
    # deploy - вебхук: крона у него нет, в `crons` он не попадает вовсе
    latest = {
        "deploy": run(1, "deploy", "done", ago=60 * 24 * 30),
        "закрытие": run(2, "закрытие", "done", ago=60),
    }
    report = silent(tmp_path, closing_ago_h=1, latest=latest)
    assert not report.quiet and "deploy" not in report.note


def test_job_never_seen_in_the_journal_waits_silently(tmp_path):
    report = silent(tmp_path, closing_ago_h=0, latest={"вайбграм": alive()})
    assert not report.quiet and not report.resumed
    assert "ещё ни разу не запускались: закрытие" in report.note


def test_irregular_schedule_is_skipped_and_named(tmp_path):
    crons = {**CRONS, "куратор": "0 2,8,12,16,20 * * *"}
    report = silent(tmp_path, closing_ago_h=1, crons=crons, latest={})
    assert report.unguarded == ("куратор",)
    assert "давностью не охраняются" in report.note and "куратор" in report.note
    assert not report.quiet


def test_the_watcher_does_not_measure_its_own_silence(tmp_path):
    latest = {
        "прогоны": run(1, "прогоны", "done", ago=60 * 24 * 5),
        "закрытие": run(2, "закрытие", "done", ago=60),
    }
    report = silent(tmp_path, closing_ago_h=1, latest=latest)
    assert not report.quiet


def test_job_removed_from_the_schedule_leaves_silence_state(tmp_path):
    told = state(tmp_path, quiet={"старый": NOW - timedelta(hours=3)})
    report = silent(tmp_path, closing_ago_h=1, runs=told)
    assert not report.resumed and "старый" not in told.quiet
    assert "сняты с давности (нет в расписании): старый" in report.note


def test_real_schedule_names_only_the_curator_unguarded(tmp_path):
    from beaver_agent.jobs import jobs

    crons = {job.name: job.cron for job in jobs if job.cron}
    report = silent(tmp_path, closing_ago_h=1, crons=crons, latest={})
    assert report.unguarded == ("куратор",)
    # Вебхуки в расписание давности не попадают
    assert {"deploy", "komodo", "t3code"}.isdisjoint(crons)


async def test_job_alarms_on_a_silent_job_and_logs_the_unguarded(
    tmp_path, monkeypatch, caplog
):
    from beaver_agent.jobs import runs as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json")
    monkeypatch.setattr(job, "BOOT", datetime.now(UTC) - timedelta(days=10))
    database = await make_db(
        tmp_path,
        [
            record("закрытие", "done", ago=60 * 50),
            *(record("прогоны", "done", ago=60 * h - 15) for h in range(1, 73)),
            record("вайбграм", "done", ago=5),
            record("deploy", "done", ago=60 * 24 * 20),
        ],
    )
    run_ = FakeRun(
        database,
        jobs=[
            ("закрытие", "20 4 * * *"),
            ("вайбграм", "*/10 * * * *"),
            ("куратор", "0 2,8,12,16,20 * * *"),
            ("deploy", None),
            ("новый", "0 3 * * *"),
        ],
    )

    with caplog.at_level("INFO", logger=job.__name__):
        await job.watch_runs(cast("Any", run_))

    (spawned,) = run_.spawned
    text = spawned[2]
    assert "ПЕРЕСТАЛИ ЗАПУСКАТЬСЯ" in text
    assert "джоб `закрытие` не запускался 2 суток при периоде 24 часа" in text
    assert "deploy" not in text and "новый" not in text
    assert "давностью не охраняются" in caplog.text and "куратор" in caplog.text
    assert "ещё ни разу не запускались: новый" in caplog.text
    assert "закрытие" in Runs.load(tmp_path / "прогоны.json").quiet
    # В те же сутки не повторяется
    await job.watch_runs(cast("Any", run_))
    assert len(run_.spawned) == 1
    await database.dispose()