import logging
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Any, cast
from urllib.parse import parse_qsl, urlsplit
from zoneinfo import ZoneInfo

from beaver_agent.hands.heartbeat import (
    PEER,
    Watch,
    alarm_text,
    beat_line,
    every_from_cron,
    look,
    parse_beat,
    parse_span,
    span,
)
from beaver_agent.hands.vibegram import PAGE, Event, Vibegram, Window

WARSAW = ZoneInfo("Europe/Warsaw")
NOW = datetime(2026, 9, 22, 15, 42, tzinfo=UTC)


def hb(at, *, every="24h", nick=PEER, eid=100):
    return Event(
        id=eid,
        nick=nick,
        kind="message",
        text=(
            f"{nick}: hb {nick} at={at.strftime('%Y-%m-%dT%H:%M:%SZ')} "
            f"boot=2026-09-01T03:00:00Z every={every} src=cron+uptime-s"
        ),
        at="09-22 07:00",
    )


def state(tmp_path, **kw):
    return Watch(path=tmp_path / "сторож.json", **kw)


def test_parse_beat_reads_fields_and_unknown_every():
    beat = parse_beat(
        "claude-kai: hb claude-kai at=2026-09-22T05:00:00Z "
        "boot=2026-09-01T03:00:00Z every=24h src=cron+uptime-s"
    )
    assert beat is not None
    assert beat.nick == "claude-kai"
    assert beat.at == datetime(2026, 9, 22, 5, 0, tzinfo=UTC)
    assert beat.boot == datetime(2026, 9, 1, 3, 0, tzinfo=UTC)
    assert beat.every == timedelta(hours=24)

    unknown = parse_beat(
        "claude-kai: hb claude-kai at=2026-09-22T05:00:00Z "
        "boot=? every=unknown src=none"
    )
    assert unknown is not None
    assert unknown.every is None and unknown.every_raw == "unknown"
    assert unknown.boot is None

    assert parse_beat("claude-kai: просто сообщение") is None
    assert parse_beat("claude-kai: ack beaver last=x", "ack") is not None


def test_span_round_trip():
    assert span(timedelta(hours=36)) == "36h"
    assert span(timedelta(minutes=90)) == "90m"
    assert span(None) == "unknown"
    assert parse_span("36h") == timedelta(hours=36)
    assert parse_span("unknown") is None
    assert parse_span("") is None


def test_every_from_cron_only_when_schedule_is_every_n_hours():
    assert every_from_cron("12 17 * * *", WARSAW, now=NOW) == timedelta(hours=24)
    assert every_from_cron("0 */6 * * *", WARSAW, now=NOW) == timedelta(hours=6)
    # Перевод стрелок не делает суточный крон 23-часовым: считаем по стенным часам
    march = datetime(2026, 3, 28, 12, tzinfo=UTC)
    assert every_from_cron("12 17 * * *", WARSAW, now=march) == timedelta(hours=24)
    # Не сводится к «раз в N часов» - и не сводим
    assert every_from_cron("0 2,8,12,16,20 * * *", WARSAW, now=NOW) is None
    assert every_from_cron("*/10 * * * *", WARSAW, now=NOW) is None
    # Недельный крон тоже сводится к N часам - это честные 168h, а не unknown
    assert every_from_cron("30 4 * * 0", WARSAW, now=NOW) == timedelta(hours=168)
    assert every_from_cron(None, WARSAW, now=NOW) is None
    assert every_from_cron("не крон", WARSAW, now=NOW) is None


def test_beat_line_says_unknown_instead_of_guessing():
    line = beat_line(
        nick="beaver",
        at=datetime(2026, 9, 22, 15, 12, tzinfo=UTC),
        boot=datetime(2026, 9, 20, 1, 2, 3, tzinfo=UTC),
        every=timedelta(hours=24),
    )
    assert line == (
        "hb beaver at=2026-09-22T15:12:00Z boot=2026-09-20T01:02:03Z every=24h src=cron"
    )
    blind = beat_line(nick="beaver", at=NOW, boot=NOW, every=None)
    assert "every=unknown src=none" in blind


def test_fresh_beat_arms_and_stays_quiet(tmp_path):
    w = state(tmp_path)
    report = look(
        w, [hb(NOW - timedelta(hours=10))], now=NOW, res=timedelta(hours=24), full=True
    )
    assert report.armed and report.alarm is None and not report.stale
    assert report.threshold == timedelta(hours=36)
    assert "armed=yes" in report.ack and "win=25" in report.ack
    assert "res=24h" in report.ack and "thr=36h" in report.ack
    assert w.seen == NOW and w.top == 100


def test_threshold_is_every_times_one_and_a_half(tmp_path):
    # 30 часов при every=24h - ещё не тревога: одна потерянная строка не в счёт
    quiet = state(tmp_path, seen=NOW - timedelta(hours=30), every=timedelta(hours=24))
    assert look(quiet, [], now=NOW, res=timedelta(hours=24), full=True).alarm is None
    # 37 часов - уже за порогом 36
    loud = state(tmp_path, seen=NOW - timedelta(hours=37), every=timedelta(hours=24))
    report = look(loud, [], now=NOW, res=timedelta(hours=24), full=True)
    assert report.alarm == "silence"
    # Порог берётся из строки соседа, а не из нашей константы
    slow = state(tmp_path, seen=NOW - timedelta(hours=37), every=timedelta(hours=48))
    assert look(slow, [], now=NOW, res=timedelta(hours=24), full=True).alarm is None


def test_last_beat_wins_over_gaps(tmp_path):
    w = state(tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24))
    report = look(
        w, [hb(NOW - timedelta(hours=1), eid=120)], now=NOW, res=None, full=True
    )
    assert report.alarm is None
    assert report.age is not None and report.age < timedelta(hours=2)
    assert "res=unknown" in report.ack


def test_unknown_every_falls_back_to_last_known_and_says_so(tmp_path):
    w = state(tmp_path, every=timedelta(hours=24))
    report = look(w, [hb(NOW, every="unknown", eid=130)], now=NOW, res=None, full=True)
    assert report.threshold == timedelta(hours=36)
    assert "thr=36h-fallback" in report.ack
    assert w.every == timedelta(hours=24)


def test_no_period_ever_means_no_measurement_and_no_alarm(tmp_path):
    w = state(tmp_path, seen=NOW - timedelta(days=9))
    report = look(w, [], now=NOW, res=timedelta(hours=24), full=True)
    assert report.alarm is None and report.threshold is None
    assert "thr=unknown" in report.ack
    assert "мерить нечем" in report.note


def test_never_seen_is_armed_no_not_alarm(tmp_path):
    w = state(tmp_path)
    report = look(w, [], now=NOW, res=timedelta(hours=24), full=True)
    assert not report.armed and report.alarm is None
    assert "armed=no" in report.ack and "last=never seen=never" in report.ack
    assert w.unarmed_since == NOW


def test_seven_days_unarmed_tells_the_human_once(tmp_path):
    w = state(tmp_path, unarmed_since=NOW - timedelta(days=7))
    assert look(w, [], now=NOW, res=timedelta(hours=24), full=True).alarm == "unarmed"
    w.unarmed_told = True
    assert look(w, [], now=NOW, res=timedelta(hours=24), full=True).alarm is None
    # Пришла первая отметка - кольцо вооружилось, счётчик забыт
    look(w, [hb(NOW, eid=140)], now=NOW, res=timedelta(hours=24), full=True)
    assert w.unarmed_since is None


def test_stale_window_forbids_alarm(tmp_path):
    w = state(
        tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24), top=500
    )
    report = look(
        w, [hb(NOW - timedelta(hours=40), eid=7)], now=NOW, res=None, full=True
    )
    assert report.stale and report.alarm is None
    assert "win=stale" in report.ack
    assert w.top == 500  # съехавшее окно не двигает отметку максимума
    # Окно снова вперёд - тревога берётся
    fresh = look(
        w, [hb(NOW - timedelta(hours=40), eid=501)], now=NOW, res=None, full=True
    )
    assert not fresh.stale


def test_alarm_not_repeated_within_a_day(tmp_path):
    w = state(
        tmp_path,
        seen=NOW - timedelta(hours=40),
        every=timedelta(hours=24),
        alerted=NOW - timedelta(hours=5),
    )
    assert look(w, [], now=NOW, res=None, full=True).alarm is None
    w.alerted = NOW - timedelta(hours=25)
    assert look(w, [], now=NOW, res=None, full=True).alarm == "silence"


def test_alarm_text_names_three_causes_and_suspects_the_channel(tmp_path):
    w = state(tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24))
    report = look(w, [], now=NOW, res=timedelta(hours=24), full=True)
    silent = alarm_text(report, mine_sent=True)
    assert "машина соседа" in silent and "крон" in silent and "канал" in silent
    assert "канал подозревать надо раньше" in silent
    # Сосед хоть чем-то отметился - канал скорее жив
    heard = look(
        state(tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24)),
        [
            Event(
                id=9,
                nick=PEER,
                kind="message",
                text=f"{PEER}: ack beaver last=x",
                at="09-22 07:00",
            )
        ],
        now=NOW,
        res=None,
        full=True,
    )
    assert heard.heard
    assert "канал скорее жив" in alarm_text(heard, mine_sent=True)


def test_state_survives_a_round_trip(tmp_path):
    w = state(
        tmp_path,
        at=NOW,
        seen=NOW,
        every=timedelta(hours=24),
        every_unknown=True,
        top=42,
        alerted=NOW,
        unarmed_since=None,
        unarmed_told=True,
    )
    w.save()
    back = Watch.load(w.path)
    assert back.at == NOW and back.seen == NOW and back.top == 42
    assert back.every == timedelta(hours=24) and back.every_unknown
    assert back.alerted == NOW and back.unarmed_told
    assert Watch.load(tmp_path / "нет.json").seen is None


class FakeRoom:
    def __init__(self, events, *, nick="beaver", send_fails=False, full=True):
        self.events = events
        self.nick = nick
        self.send_fails = send_fails
        self.full = full
        self.said = []
        self.limit = None

    async def window(self, limit=25):
        self.limit = limit
        return Window(events=self.events, full=self.full)

    async def say(self, text):
        if self.send_fails:
            msg = "хаб молчит"
            raise RuntimeError(msg)
        self.said.append(text)


class FakeRun:
    def __init__(self, cron="42 */6 * * *"):
        self.job = SimpleNamespace(cron=cron, name="сторож")
        self.scheduler = SimpleNamespace(tz=WARSAW)
        self.spawned = []

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


async def test_watch_job_prints_ack_and_spawns_alarm(tmp_path, monkeypatch):
    from beaver_agent.jobs import heartbeat as job

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "сторож.json")
    Watch(
        path=tmp_path / "сторож.json",
        seen=datetime.now(UTC) - timedelta(hours=40),
        every=timedelta(hours=24),
    ).save()
    room = FakeRoom([])
    monkeypatch.setattr(job, "VIBEGRAM_HB", room)
    run = FakeRun()

    await job.watch(cast("Any", run))

    assert room.limit == 25  # окно запрошено явным размером
    assert room.said and room.said[0].startswith(f"ack {PEER} ")
    assert "res=6h" in room.said[0]
    assert len(run.spawned) == 1
    agent, title, text = run.spawned[0]
    assert agent == "beaver-triage" and title == "сторож"
    assert "канал подозревать надо раньше" in text
    # Тревога записана в состояние - следующий прогон в те же сутки молчит
    assert Watch.load(tmp_path / "сторож.json").alerted is not None
    await job.watch(cast("Any", run))
    assert len(run.spawned) == 1


async def test_beat_job_sends_period_from_its_own_cron(monkeypatch):
    from beaver_agent.jobs import heartbeat as job

    room = FakeRoom([], nick="beaver")
    monkeypatch.setattr(job, "VIBEGRAM_HB", room)
    await job.beat(cast("Any", FakeRun(cron="12 17 * * *")))
    assert room.said[0].startswith("hb beaver at=")
    assert "every=24h src=cron" in room.said[0]

    odd = FakeRoom([], nick="beaver")
    monkeypatch.setattr(job, "VIBEGRAM_HB", odd)
    await job.beat(cast("Any", FakeRun(cron="0 2,8,12,16,20 * * *")))
    assert "every=unknown src=none" in odd.said[0]


async def test_watch_job_keeps_quiet_when_the_room_is_unreadable(tmp_path, monkeypatch):
    from beaver_agent.jobs import heartbeat as job

    class Dead(FakeRoom):
        async def window(self, limit=25):
            msg = "хаб не отвечает"
            raise RuntimeError(msg)

    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "сторож.json")
    monkeypatch.setattr(job, "VIBEGRAM_HB", Dead([]))
    run = FakeRun()
    await job.watch(cast("Any", run))
    assert not run.spawned
    assert not (tmp_path / "сторож.json").exists()


class FakeHub:
    """Хаб как он есть: id сквозные по ВСЕМ комнатам, `/api/state` - наша.

    Хвост комнаты лежит за `/api/state?view=…&after=<id>` и отдаётся
    страницами вперёд; `/api/pending` - курсор, и сторожу в него ходить
    нечего: то же событие второй раз оттуда не придёт.
    """

    def __init__(self, events, *, page=PAGE):
        self.events = events
        self.page = page
        self.said = []
        self.afters = []

    async def call(self, method, path, body):
        if method == "POST" and path == "/api/messages":
            self.said.append(body["body"])
            return {"ok": True}
        if path == "/api/room":
            return {"room": {"id": "dbb", "name": "kai-rem-watch", "viewToken": "v-1"}}
        if path.startswith("/api/state"):
            query = dict(parse_qsl(urlsplit(path).query))
            assert query["view"] == "v-1"
            after = int(query.get("after", 0))
            self.afters.append(after)
            rest = [e for e in self.events if e["id"] > after]
            return {"room": {}, "agents": [], "events": rest[: self.page]}
        msg = f"сторож не должен ходить в {path}"
        raise AssertionError(msg)


def raw(eid, nick, kind, body=None, at="2026-09-23T05:00:02.000Z"):
    return {
        "id": eid,
        "roomId": "dbb",
        "nick": nick,
        "kind": kind,
        "payload": {} if body is None else {"body": body},
        "createdAt": at,
    }


def room(monkeypatch, events, *, page=PAGE):
    hub = FakeHub(events, page=page)

    async def call(self, method, path, body):
        return await hub.call(method, path, body)

    monkeypatch.setattr(Vibegram, "_call", call)
    return hub, Vibegram(hub="http://hub", token="t", nick="beaver-rem")


KAI_HB = (
    f"hb {PEER} at=2026-09-23T05:00:01Z boot=2026-09-14T08:21:55Z "
    "every=24h src=cron+uptime-s"
)


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

    # В комнате три события, а по хабу за то же время прошло три десятка:
    # номера сквозные, и между 296 и 336 лежат чужие комнаты. Отметка соседа
    # (#329) обязана быть в окне - в комнате она первая с конца.
    hub, client = room(
        monkeypatch,
        [
            raw(296, PEER, "agent_join"),
            raw(329, PEER, "message", KAI_HB),
            raw(331, "beaver-rem", "agent_leave"),
            raw(332, "beaver-rem", "message", "hb beaver-rem at=2026-09-23T15:12:01Z"),
            raw(336, PEER, "message", "ack beaver-rem last=2026-09-23T15:12:01Z"),
        ],
    )
    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "сторож.json")
    monkeypatch.setattr(job, "VIBEGRAM_HB", client)

    window = await client.window(25)
    assert window.full
    assert [e.id for e in window.events] == [329, 332, 336]  # join/leave не в счёт
    assert parse_beat(window.events[0].text) is not None

    # Два прогона подряд: отметка соседа в окне оба раза, а не только в первом
    with caplog.at_level(logging.INFO, logger="beaver_agent.jobs.heartbeat"):
        await job.watch(cast("Any", FakeRun()))
        await job.watch(cast("Any", FakeRun()))
    beats = [m for m in caplog.messages if "отметок соседа в окне: 1" in m]
    assert len(beats) == 2
    assert all("история дочитана" in m for m in beats)
    assert all("last=2026-09-23T05:00:01Z" in a and "armed=yes" in a for a in hub.said)
    assert all("win=25 " in a for a in hub.said)
    assert hub.afters == [0, 0, 0]  # одна страница на прогон, читать дальше нечего


async def test_a_hundred_noise_lines_do_not_push_the_control_beat_out(
    tmp_path, monkeypatch
):
    from beaver_agent.jobs import heartbeat as job

    # Учения соседа: сотня шумовых строк в комнату и контрольная hb после них.
    noise = [raw(400 + i, PEER, "message", f"шум {i}") for i in range(100)]
    hub, client = room(monkeypatch, [*noise, raw(600, PEER, "message", KAI_HB)])
    monkeypatch.setattr(job, "STATE_FILE", tmp_path / "сторож.json")
    monkeypatch.setattr(job, "VIBEGRAM_HB", client)

    await job.watch(cast("Any", FakeRun()))

    assert "last=2026-09-23T05:00:01Z" in hub.said[0] and "armed=yes" in hub.said[0]
    assert Watch.load(tmp_path / "сторож.json").top == 600


def test_unread_window_is_named_and_takes_no_alarm(tmp_path):
    late = {"seen": NOW - timedelta(hours=40), "every": timedelta(hours=24)}
    partial = look(state(tmp_path, **late), [], now=NOW, res=None, full=False)
    assert not partial.full and partial.alarm is None
    assert "win=25/partial" in partial.ack
    assert "окно не дочитано" in partial.note
    # То же пустое окно, но дочитанное, - это показание, и оно даёт тревогу
    whole = look(state(tmp_path, **late), [], now=NOW, res=None, full=True)
    assert whole.alarm == "silence" and "win=25 " in whole.ack
    assert "сообщений комнаты: 0 из 25, история дочитана" in whole.note