"""The generalized change watcher (#73618): cheap on-disk signatures →
``pet.changed`` / ``cron.changed`` / ``sessions.changed`` global broadcasts.

Behavior contracts, exercised against a real temp HERMES_HOME (no mocks on the
filesystem path): first sighting seeds silently, a moved signature broadcasts
once, the sessions floor coalesces a write burst but keeps its trailing edge,
and the pet signature only moves for a *renderable* pet.
"""

import os
import sqlite3
import time

import pytest

from tui_gateway import server


@pytest.fixture()
def watcher_home(tmp_path, monkeypatch):
    (tmp_path / "config.yaml").write_text("display: {}\n")
    (tmp_path / "cron").mkdir()

    monkeypatch.setattr(server, "_hermes_home", str(tmp_path))
    monkeypatch.setattr(server, "_cfg_cache", None)
    monkeypatch.setattr(server, "_change_sigs", {})
    monkeypatch.setattr(server, "_change_checked_at", {})
    monkeypatch.setattr(server, "_change_broadcast_at", {})
    monkeypatch.setattr(server, "_bot_relay_outbox_seen", 0)
    monkeypatch.setattr(server, "_pairing_roots_cache", None, raising=False)
    monkeypatch.setattr(server, "_sessions_db_sig_cache", {})

    events = []
    monkeypatch.setattr(
        server, "_broadcast_global_event", lambda ev, payload=None: events.append((ev, payload))
    )
    return tmp_path, events


def _write_session_change(db_path, title):
    conn = sqlite3.connect(db_path)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS sessions "
        "(id TEXT PRIMARY KEY, title TEXT, last_activity_at REAL, message_count INTEGER)"
    )
    conn.execute(
        "INSERT INTO sessions(id, title, last_activity_at, message_count) VALUES ('s1', ?, 1, 1) "
        "ON CONFLICT(id) DO UPDATE SET title = excluded.title",
        (title,),
    )
    conn.commit()
    conn.close()


def test_first_sighting_seeds_without_broadcasting(watcher_home):
    home, events = watcher_home
    (home / "cron" / "jobs.json").write_text("[]")
    (home / "state.db").write_text("x")

    server._broadcast_watched_changes(now=0.0)

    assert events == []


def test_cron_jobs_file_move_broadcasts_cron_changed(watcher_home):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    (home / "cron" / "jobs.json").write_text("[]")
    server._broadcast_watched_changes(now=10.0)

    assert ("cron.changed", {}) in events


def test_state_db_move_broadcasts_sessions_changed(watcher_home):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    _write_session_change(home / "state.db", "created")
    server._broadcast_watched_changes(now=10.0)

    assert ("sessions.changed", {}) in events


def _seed_store(db_path):
    conn = sqlite3.connect(db_path)
    conn.executescript(
        "CREATE TABLE sessions (id TEXT PRIMARY KEY, title TEXT, started_at REAL, "
        "message_count INTEGER, last_activity_at REAL, last_activity_description TEXT);"
        "CREATE TABLE gateway_heartbeats (backend_id TEXT PRIMARY KEY, last_heartbeat REAL);"
        "INSERT INTO sessions VALUES ('s1', 'hello', 1, 3, 1, NULL);"
        "INSERT INTO gateway_heartbeats VALUES ('backend-1', 1);"
    )
    conn.commit()
    conn.close()


def _write(db_path, sql, *params):
    conn = sqlite3.connect(db_path)
    conn.execute(sql, params)
    conn.commit()
    conn.close()


def test_activity_heartbeats_do_not_broadcast_sessions_changed(watcher_home):
    """#98005: the gateway heartbeat and the session activity stamp rewrite state.db every
    minute with no session change. Each used to fire sessions.changed, so the Desktop
    Sessions panel refreshed on its own every heartbeat window."""
    home, events = watcher_home
    db = home / "state.db"
    _seed_store(db)
    server._broadcast_watched_changes(now=0.0)

    for tick in range(1, 4):
        time.sleep(0.02)
        _write(db, "UPDATE gateway_heartbeats SET last_heartbeat = ?", tick * 60.0)
        _write(db, "UPDATE sessions SET last_activity_at = ?, last_activity_description = ? "
                   "WHERE id = 's1'", tick * 60.0, f"tool {tick}")
        server._broadcast_watched_changes(now=tick * 10.0)

    assert ("sessions.changed", {}) not in events


@pytest.mark.parametrize("sql", [
    "INSERT INTO sessions VALUES ('s2', 'new', 2, 0, 2, NULL)",
    "UPDATE sessions SET title = 'renamed' WHERE id = 's1'",
    "UPDATE sessions SET message_count = 4 WHERE id = 's1'",
    "DELETE FROM sessions WHERE id = 's1'",
])
def test_session_row_changes_broadcast_sessions_changed(watcher_home, sql):
    home, events = watcher_home
    db = home / "state.db"
    _seed_store(db)
    server._broadcast_watched_changes(now=0.0)

    time.sleep(0.02)
    _write(db, sql)
    server._broadcast_watched_changes(now=10.0)

    assert ("sessions.changed", {}) in events


def test_unreadable_store_keeps_last_digest(watcher_home):
    """A locked/unreadable moment after a good read must not flip to the mtime signature:
    digest -> mtime -> digest would broadcast twice for nothing."""
    home, events = watcher_home
    db = home / "state.db"
    _seed_store(db)
    server._broadcast_watched_changes(now=0.0)

    db.write_text("not-sqlite")
    server._broadcast_watched_changes(now=10.0)

    assert ("sessions.changed", {}) not in events


def test_projects_db_move_broadcasts_projects_changed(watcher_home):
    """#53046 / #56757: the CLI and other windows write projects.db directly, in
    processes that never touch this gateway's transports. Without a watch, a
    `hermes projects create` (or a set_primary / folder edit from another
    window) leaves the Desktop's project tree stale until an unrelated refresh."""
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    (home / "projects.db").write_text("x")
    server._broadcast_watched_changes(now=10.0)

    assert ("projects.changed", {}) in events


def test_served_profile_projects_db_move_broadcasts_projects_changed(watcher_home, monkeypatch):
    """A backend serving a sibling profile watches that profile's projects.db too —
    the sibling-home half of the sessions.changed contract (#53046)."""
    home, events = watcher_home
    coder_home = home / "profiles" / "coder"
    coder_home.mkdir(parents=True)
    monkeypatch.setattr(server, "_served_profile_homes", set())
    monkeypatch.setattr("hermes_cli.profiles.get_profile_dir", lambda name: home / "profiles" / name)
    assert server._profile_home("coder") == coder_home
    server._broadcast_watched_changes(now=0.0)

    (coder_home / "projects.db").write_text("x")
    server._broadcast_watched_changes(now=10.0)

    assert ("projects.changed", {}) in events


def test_projects_sig_does_not_track_state_db_writes(watcher_home):
    """A state.db move must not fire projects.changed — the two stores are
    independent and their consumers refetch different surfaces."""
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    (home / "state.db").write_text("x")
    server._broadcast_watched_changes(now=10.0)

    assert ("projects.changed", {}) not in events


def test_served_profile_store_move_broadcasts_sessions_changed(watcher_home, monkeypatch):
    """A backend serving a sibling profile must see that profile's state.db
    move too — otherwise a routed profile's Bot Chat never refreshes (#99333)."""
    home, events = watcher_home
    bot_home = home / "profiles" / "bot"
    bot_home.mkdir(parents=True)
    monkeypatch.setattr(server, "_served_profile_homes", set())
    monkeypatch.setattr("hermes_cli.profiles.get_profile_dir", lambda name: home / "profiles" / name)
    assert server._profile_home("bot") == bot_home
    server._broadcast_watched_changes(now=0.0)

    _write_session_change(bot_home / "state.db", "created")
    server._broadcast_watched_changes(now=10.0)

    assert ("sessions.changed", {}) in events


def test_gateway_state_move_broadcasts_platforms_changed(watcher_home):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    (home / "gateway_state.json").write_text('{"platforms": {}}')
    server._broadcast_watched_changes(now=10.0)

    assert ("platforms.changed", {}) in events


def test_pending_pairing_request_broadcasts_pairing_changed(watcher_home):
    """A new pending request must reach the Messaging page on its own signal.

    The messaging gateway writes the pending code from a different process, and
    it moves nothing in gateway_state.json — so platforms.changed cannot stand
    in for this. Without a dedicated signal the badge stays invisible until an
    unrelated connect/disconnect happens to fire.
    """
    home, events = watcher_home
    store = home / "platforms" / "pairing"
    store.mkdir(parents=True)
    server._broadcast_watched_changes(now=0.0)

    (store / "telegram-pending.json").write_text('{"abc": {"user_id": "1"}}')
    server._broadcast_watched_changes(now=10.0)

    assert ("pairing.changed", {}) in events
    assert ("platforms.changed", {}) not in events


def test_pairing_signal_follows_a_profile_store(watcher_home):
    """Each profile keeps its own whitelist, and the page can be scoped to any."""
    home, events = watcher_home
    store = home / "profiles" / "work" / "platforms" / "pairing"
    store.mkdir(parents=True)
    (home / "profiles" / "work" / "config.yaml").write_text("{}\n")  # identity marker: a bare dir is not a profile
    server._broadcast_watched_changes(now=0.0)

    (store / "telegram-approved.json").write_text('{"u1": {"user_id": "u1"}}')
    server._broadcast_watched_changes(now=10.0)

    assert ("pairing.changed", {}) in events


def test_pairing_probe_reuses_live_profile_roots_until_the_profile_set_moves(watcher_home, monkeypatch):
    """The per-profile liveness probe (~14 stats each) runs once per profiles/ mtime + TTL, not
    on every 2 s tick (#114041 §2); ledger writes under known roots are still seen each tick,
    and a newly created profile is picked up because creating it bumps the parent's mtime."""
    import hermes_constants

    home, events = watcher_home
    live_calls = []
    real_live = hermes_constants.named_profile_is_live
    monkeypatch.setattr(hermes_constants, "named_profile_is_live",
                        lambda p: live_calls.append(p.name) or real_live(p))

    def _profile(name):
        (home / "profiles" / name / "platforms" / "pairing").mkdir(parents=True)
        (home / "profiles" / name / "config.yaml").write_text("{}\n", encoding="utf-8")

    _profile("work")
    server._broadcast_watched_changes(now=0.0)
    (home / "profiles" / "work" / "platforms" / "pairing" / "telegram-pending.json").write_text("{}", encoding="utf-8")
    server._broadcast_watched_changes(now=10.0)
    assert events == [("pairing.changed", {})]
    assert live_calls == ["work"]  # second tick reused the cached roots, still saw the ledger

    _profile("play")
    os.utime(home / "profiles", ns=(0, 10**18))  # deterministic parent-mtime bump
    server._broadcast_watched_changes(now=20.0)
    (home / "profiles" / "play" / "platforms" / "pairing" / "discord-approved.json").write_text("{}", encoding="utf-8")
    server._broadcast_watched_changes(now=30.0)
    assert events == [("pairing.changed", {})] * 2
    assert sorted(live_calls) == ["play", "work", "work"]


def test_rate_limit_churn_does_not_broadcast_pairing_changed(watcher_home):
    """_rate_limits.json moves on every unauthorized DM, including ones that
    produce no new row — signalling on it would refetch for nothing."""
    home, events = watcher_home
    store = home / "platforms" / "pairing"
    store.mkdir(parents=True)
    (store / "telegram-pending.json").write_text("{}")
    server._broadcast_watched_changes(now=0.0)

    (store / "_rate_limits.json").write_text('{"telegram:1": 123}')
    server._broadcast_watched_changes(now=10.0)

    assert ("pairing.changed", {}) not in events


def test_sessions_floor_coalesces_burst_but_keeps_trailing_edge(watcher_home):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    _write_session_change(home / "state.db", "first")
    server._broadcast_watched_changes(now=10.0)
    events.clear()

    # A second write lands inside the 2s floor: no broadcast yet…
    time.sleep(0.02)
    _write_session_change(home / "state.db", "second")
    server._broadcast_watched_changes(now=11.0)
    assert events == []

    # …but the change is not lost — it fires once the window opens.
    server._broadcast_watched_changes(now=13.0)
    assert ("sessions.changed", {}) in events


def test_pet_sig_stays_off_without_a_renderable_pet(watcher_home):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    # Config flips enabled but no pet exists on disk → signature stays ("off",).
    (home / "config.yaml").write_text("display:\n  pet:\n    enabled: true\n    slug: boba\n")
    server._cfg_cache = None
    server._broadcast_watched_changes(now=10.0)

    assert not [e for e in events if e[0] == "pet.changed"]


def test_renderable_pet_broadcasts_meta_payload(watcher_home, monkeypatch):
    home, events = watcher_home
    (home / "config.yaml").write_text("display:\n  pet:\n    enabled: true\n    slug: boba\n")
    server._cfg_cache = None
    server._broadcast_watched_changes(now=0.0)

    sheet = home / "sheet.png"
    sheet.write_text("png")

    class FakePet:
        slug = "boba"
        display_name = "Boba"
        exists = True
        spritesheet = sheet

    monkeypatch.setattr(server, "_pet_active_selection", lambda: (True, FakePet(), 0.33))
    server._broadcast_watched_changes(now=10.0)

    pet_events = [e for e in events if e[0] == "pet.changed"]
    assert pet_events
    payload = pet_events[0][1]
    assert payload["enabled"] is True
    assert payload["slug"] == "boba"
    assert payload["spritesheetRevision"]


def test_enqueued_envelope_broadcasts_outbox_pending(watcher_home):
    """A cross-connection envelope written by the agent process must reach the
    Desktop's push-triggered drain on its own signal (#93091) — the drain poll
    is the backstop, not the transport."""
    home, events = watcher_home
    outbox = home / "bot_relay" / "outbox"
    outbox.mkdir(parents=True)
    server._broadcast_watched_changes(now=0.0)

    (outbox / ("a" * 32 + ".json")).write_text('{"id": "' + "a" * 32 + '"}')
    server._broadcast_watched_changes(now=10.0)

    assert ("bot_relay.outbox.pending", {}) in events


def test_drained_outbox_does_not_rebroadcast_pending(watcher_home):
    """Signature is monotone: a drain empties outbox/ (rename → claimed/), and
    that emptying must NOT look like a change — only new envelopes fire."""
    home, events = watcher_home
    outbox = home / "bot_relay" / "outbox"
    outbox.mkdir(parents=True)
    envelope = outbox / ("b" * 32 + ".json")
    envelope.write_text("{}")
    server._broadcast_watched_changes(now=0.0)

    envelope.unlink()  # the Desktop drained it
    server._broadcast_watched_changes(now=10.0)
    server._broadcast_watched_changes(now=20.0)

    assert not [e for e in events if e[0] == "bot_relay.outbox.pending"]


def test_new_envelope_after_drain_fires_pending_again(watcher_home):
    """The other half of the monotone contract: the watermark must not eat
    GENUINELY new envelopes. write → drain → write-newer fires twice."""
    home, events = watcher_home
    outbox = home / "bot_relay" / "outbox"
    outbox.mkdir(parents=True)
    first = outbox / ("c" * 32 + ".json")
    first.write_text("{}")
    server._broadcast_watched_changes(now=0.0)
    first.write_text("{}")  # make the first sighting a change, not a seed
    bump_ns = first.stat().st_mtime_ns + 1_000_000
    os.utime(first, ns=(bump_ns, bump_ns))  # strictly newer, FS-independent
    server._broadcast_watched_changes(now=10.0)

    first.unlink()  # the Desktop drained it
    server._broadcast_watched_changes(now=20.0)

    second = outbox / ("d" * 32 + ".json")
    second.write_text("{}")
    newer_ns = bump_ns + 1_000_000  # strictly beyond the watermark
    os.utime(second, ns=(newer_ns, newer_ns))
    server._broadcast_watched_changes(now=30.0)

    assert [e for e in events if e[0] == "bot_relay.outbox.pending"] == [
        ("bot_relay.outbox.pending", {}),
        ("bot_relay.outbox.pending", {}),
    ]


def test_no_outbox_dir_never_fires_pending(watcher_home):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)
    server._broadcast_watched_changes(now=10.0)

    assert not [e for e in events if e[0] == "bot_relay.outbox.pending"]


def test_broken_probe_never_kills_the_pass(watcher_home, monkeypatch):
    home, events = watcher_home
    server._broadcast_watched_changes(now=0.0)

    monkeypatch.setitem(
        server._CHANGE_WATCHES,
        "cron.changed",
        (1.0, lambda: (_ for _ in ()).throw(RuntimeError("boom")), lambda: {}),
    )
    _write_session_change(home / "state.db", "created")
    server._broadcast_watched_changes(now=10.0)

    # The broken cron probe is skipped; sessions still broadcasts.
    assert ("sessions.changed", {}) in events