"""Tests for gateway session stall watchdog (#72016 item 2)."""

from __future__ import annotations

import asyncio
import time
from types import SimpleNamespace

import pytest

from agent.session_activity import ActivityProvenance, build_activity_snapshot
from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
from gateway.session_stall import (
    format_session_stall_notification,
    resolve_session_idle_seconds_from_activity,
    should_clear_session_stall_notification,
    should_emit_session_stall_notification,
)


class _FakeAdapter:
    def __init__(self):
        self._pending_messages = {}
        self.sent = []

    async def send(self, chat_id, content, metadata=None):
        self.sent.append(
            {"chat_id": chat_id, "content": content, "metadata": metadata}
        )


class _FakeAgent:
    """Exposes the shared #72039 activity snapshot as the sole progress source."""

    def __init__(
        self,
        last_activity_ts: float,
        *,
        description: str = "api call",
        provenance: ActivityProvenance = ActivityProvenance.UNKNOWN,
    ):
        self._last_activity_ts = last_activity_ts
        self._last_activity_desc = description
        self._last_activity_provenance = provenance

    def get_activity_summary(self):
        return build_activity_snapshot(
            last_activity_at=self._last_activity_ts,
            last_activity_description=self._last_activity_desc,
            last_activity_provenance=self._last_activity_provenance,
        )


class _AgentWithoutSummary:
    """Agent with a raw clock but no shared summary consumer API."""

    def __init__(self, last_activity_ts: float):
        self._last_activity_ts = last_activity_ts


def test_should_emit_requires_pending_and_idle():
    assert should_emit_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=400,
        has_pending_inbound=True,
        already_notified=False,
    )
    assert not should_emit_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=400,
        has_pending_inbound=False,
        already_notified=False,
    )
    assert not should_emit_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=100,
        has_pending_inbound=True,
        already_notified=False,
    )
    assert not should_emit_session_stall_notification(
        timeout_seconds=0,
        idle_seconds=9999,
        has_pending_inbound=True,
        already_notified=False,
    )
    assert not should_emit_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=400,
        has_pending_inbound=True,
        already_notified=True,
    )


def test_should_clear_when_pending_gone_or_activity_resumes():
    assert should_clear_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=400,
        has_pending_inbound=False,
    )
    assert should_clear_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=10,
        has_pending_inbound=True,
    )
    assert not should_clear_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=400,
        has_pending_inbound=True,
    )


def test_should_clear_holds_latch_when_idle_unknown():
    assert not should_clear_session_stall_notification(
        timeout_seconds=300,
        idle_seconds=None,
        has_pending_inbound=True,
    )


def test_format_session_stall_notification_minutes():
    msg = format_session_stall_notification(125)
    assert "2 min ago" in msg
    assert "/new" in msg
    assert format_session_stall_notification(30).count("1 min ago") == 1


def test_resolve_idle_uses_shared_activity_snapshot_only():
    now = 1_000_000.0
    snap = build_activity_snapshot(
        last_activity_at=now - 120,
        last_activity_description="tool: terminal",
        last_activity_provenance=ActivityProvenance.UNKNOWN,
        now=now,
    )
    assert resolve_session_idle_seconds_from_activity(snap, now=now) == 120.0
    assert resolve_session_idle_seconds_from_activity(None, now=now) is None
    assert resolve_session_idle_seconds_from_activity({}, now=now) is None


def test_resolve_idle_prefers_seconds_since_activity_field():
    idle = resolve_session_idle_seconds_from_activity(
        {
            "seconds_since_activity": 42.5,
            "last_activity_at": 1.0,  # must be ignored when seconds present
        },
        now=999.0,
    )
    assert idle == 42.5


def _runner_for_stall(adapter: _FakeAdapter) -> GatewayRunner:
    r = GatewayRunner.__new__(GatewayRunner)
    r._running = True
    r.adapters = {"fake": adapter}
    r._profile_adapters = {}
    r._running_agents = {}
    r._running_agents_ts = {}
    r._queued_events = {}
    r._session_stall_notified = {}
    r._thread_metadata_for_source = lambda source, *a, **k: {
        "thread_id": getattr(source, "thread_id", None)
    }
    return r


def _pending_event(chat_id: str = "chat-1", thread_id: str | None = None):
    source = SimpleNamespace(chat_id=chat_id, thread_id=thread_id, platform=None)
    return SimpleNamespace(text="follow-up", source=source, timestamp=time.time())


@pytest.mark.asyncio
async def test_check_session_stalls_notifies_once(monkeypatch):
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    monkeypatch.setenv("HERMES_SESSION_STALL_TIMEOUT", "60")
    session_key = "agent:main:telegram:dm:1"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)

    sent = await runner._check_session_stalls(60)
    assert sent == 1
    assert len(adapter.sent) == 1
    assert "/new" in adapter.sent[0]["content"]
    assert runner._session_stall_notified.get(session_key) is True

    # Second pass must not spam.
    sent2 = await runner._check_session_stalls(60)
    assert sent2 == 0
    assert len(adapter.sent) == 1


@pytest.mark.asyncio
async def test_check_session_stalls_skips_fresh_activity():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:2"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 5)

    sent = await runner._check_session_stalls(60)
    assert sent == 0
    assert adapter.sent == []


@pytest.mark.asyncio
async def test_check_session_stalls_skips_without_pending():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:3"
    runner._running_agents[session_key] = _FakeAgent(time.time() - 999)

    sent = await runner._check_session_stalls(60)
    assert sent == 0
    assert adapter.sent == []


@pytest.mark.asyncio
async def test_check_session_stalls_clears_latch_when_pending_drains():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:4"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 200)

    assert await runner._check_session_stalls(60) == 1
    assert session_key in runner._session_stall_notified

    adapter._pending_messages.clear()
    assert await runner._check_session_stalls(60) == 0
    assert session_key not in runner._session_stall_notified


@pytest.mark.asyncio
async def test_check_session_stalls_skips_pending_sentinel_without_activity():
    """Pending construction has no shared activity snapshot — no parallel clocks."""
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:5"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _AGENT_PENDING_SENTINEL
    runner._running_agents_ts[session_key] = time.time() - 90

    sent = await runner._check_session_stalls(60)
    assert sent == 0
    assert adapter.sent == []


@pytest.mark.asyncio
async def test_check_session_stalls_ignores_raw_clock_without_summary():
    """Do not fall back to agent._last_activity_ts outside get_activity_summary()."""
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:raw"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _AgentWithoutSummary(time.time() - 999)
    runner._running_agents_ts[session_key] = time.time() - 999

    sent = await runner._check_session_stalls(60)
    assert sent == 0
    assert adapter.sent == []


@pytest.mark.asyncio
async def test_session_stall_watcher_disabled_when_timeout_zero(monkeypatch):
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    monkeypatch.setenv("HERMES_SESSION_STALL_TIMEOUT", "0")
    session_key = "agent:main:telegram:dm:6"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 999)

    task = asyncio.create_task(runner._session_stall_watcher(interval=1))
    await asyncio.sleep(0.05)
    # Force one check path via timeout read; watcher should no-op when 0.
    assert runner._session_stall_timeout_seconds() == 0.0
    runner._running = False
    await asyncio.wait_for(task, timeout=2)
    assert adapter.sent == []


@pytest.mark.asyncio
async def test_check_session_stalls_queued_events_overflow_notifies():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:overflow"
    event = _pending_event()
    runner._queued_events[session_key] = [event]
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
    runner._adapter_for_source = lambda source: adapter

    assert await runner._check_session_stalls(60) == 1
    assert adapter.sent and "/new" in adapter.sent[0]["content"]


@pytest.mark.asyncio
async def test_check_session_stalls_scans_profile_adapters():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(_FakeAdapter())  # empty primary path unused
    runner.adapters = {}
    runner._profile_adapters = {"coder": {"fake": adapter}}
    session_key = "agent:coder:telegram:dm:1"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)

    assert await runner._check_session_stalls(60) == 1
    assert len(adapter.sent) == 1


@pytest.mark.asyncio
async def test_check_session_stalls_logs_compression_provenance(caplog):
    """Stale compression stamps still stall, but provenance stays visible."""
    import logging

    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:compress"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(
        time.time() - 120,
        description="compressing context",
        provenance=ActivityProvenance.AGENT_COMPRESSION,
    )

    with caplog.at_level(logging.WARNING):
        assert await runner._check_session_stalls(60) == 1
    assert any("agent.compression" in r.message for r in caplog.records)
    assert any("compressing context" in r.message for r in caplog.records)


@pytest.mark.asyncio
async def test_check_session_stalls_skips_active_compression_heartbeat():
    """Fresh agent.compression heartbeats are progress, not a silent stall."""
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:compacting"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(
        time.time() - 5,
        description="context compression in progress",
        provenance=ActivityProvenance.AGENT_COMPRESSION,
    )

    assert await runner._check_session_stalls(60) == 0
    assert adapter.sent == []
    assert session_key not in runner._session_stall_notified


@pytest.mark.asyncio
async def test_check_session_stalls_does_not_renotify_after_summary_gap():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:gap"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)

    assert await runner._check_session_stalls(60) == 1

    # Transient observation gap (no summary API).
    runner._running_agents[session_key] = _AgentWithoutSummary(time.time() - 999)
    assert await runner._check_session_stalls(60) == 0
    assert runner._session_stall_notified.get(session_key) is True

    # Stale progress returns — must not spam again in the same episode.
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
    assert await runner._check_session_stalls(60) == 0
    assert len(adapter.sent) == 1


@pytest.mark.asyncio
async def test_check_session_stalls_retries_after_send_failure():
    class _FailThenOk(_FakeAdapter):
        def __init__(self):
            super().__init__()
            self.calls = 0

        async def send(self, chat_id, content, metadata=None):
            self.calls += 1
            if self.calls == 1:
                raise RuntimeError("boom")
            await super().send(chat_id, content, metadata=metadata)

    adapter = _FailThenOk()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:retry"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)

    assert await runner._check_session_stalls(60) == 0
    assert session_key not in runner._session_stall_notified
    assert await runner._check_session_stalls(60) == 1
    assert runner._session_stall_notified.get(session_key) is True
    assert len(adapter.sent) == 1


@pytest.mark.asyncio
async def test_check_session_stalls_retries_after_soft_send_failure():
    class _SoftFailThenOk(_FakeAdapter):
        def __init__(self):
            super().__init__()
            self.calls = 0

        async def send(self, chat_id, content, metadata=None):
            from gateway.platforms.base import SendResult

            self.calls += 1
            if self.calls == 1:
                return SendResult(success=False, error="chat not found")
            await super().send(chat_id, content, metadata=metadata)
            return SendResult(success=True)

    adapter = _SoftFailThenOk()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:soft"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)

    assert await runner._check_session_stalls(60) == 0
    assert session_key not in runner._session_stall_notified
    assert await runner._check_session_stalls(60) == 1
    assert runner._session_stall_notified.get(session_key) is True
    assert len(adapter.sent) == 1


@pytest.mark.asyncio
async def test_check_session_stalls_renotifies_after_resume_then_restall():
    adapter = _FakeAdapter()
    runner = _runner_for_stall(adapter)
    session_key = "agent:main:telegram:dm:episode"
    adapter._pending_messages[session_key] = _pending_event()
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)

    assert await runner._check_session_stalls(60) == 1

    # Activity resumes (still pending) — clears latch for a new episode.
    runner._running_agents[session_key] = _FakeAgent(time.time() - 5)
    assert await runner._check_session_stalls(60) == 0
    assert session_key not in runner._session_stall_notified

    # Stall again — second episode may notify once more.
    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
    assert await runner._check_session_stalls(60) == 1
    assert len(adapter.sent) == 2


def test_resolve_idle_rejects_nonfinite_seconds_since_activity():
    now = 1_000_000.0
    idle = resolve_session_idle_seconds_from_activity(
        {
            "seconds_since_activity": float("nan"),
            "last_activity_at": now - 15,
        },
        now=now,
    )
    assert idle == 15.0


def test_session_stall_timeout_in_default_config():
    from hermes_cli.config import DEFAULT_CONFIG

    timeout = DEFAULT_CONFIG["agent"]["session_stall_timeout"]
    assert isinstance(timeout, (int, float))
    assert timeout > 0  # enabled by default; 0 would disable the watchdog


class _NeverResolvingAdapter:
    """Adapter whose send() hangs forever (wedged transport)."""

    def __init__(self):
        self._pending_messages = {}
        self.send_attempts = 0

    async def send(self, chat_id, content, metadata=None):
        self.send_attempts += 1
        await asyncio.Event().wait()  # never resolves


@pytest.mark.asyncio
async def test_check_session_stalls_bounds_wedged_send(monkeypatch):
    """Round-2 #2: a never-resolving adapter.send must not wedge the watcher.

    The bounded send times out, does NOT latch (retry next tick), the pass
    completes so the watcher keeps ticking, and a healthy sibling candidate
    still receives its notification in the same pass.
    """
    import gateway.run as gateway_run

    monkeypatch.setattr(
        gateway_run, "_STALL_NOTIFY_SEND_TIMEOUT_SECONDS", 0.1
    )
    wedged = _NeverResolvingAdapter()
    healthy = _FakeAdapter()
    runner = _runner_for_stall(wedged)
    runner.adapters = {"wedged": wedged, "healthy": healthy}

    wedged_key = "agent:main:telegram:dm:wedged"
    healthy_key = "agent:main:discord:dm:healthy"
    wedged._pending_messages[wedged_key] = _pending_event(chat_id="chat-w")
    healthy._pending_messages[healthy_key] = _pending_event(chat_id="chat-h")
    runner._running_agents[wedged_key] = _FakeAgent(time.time() - 120)
    runner._running_agents[healthy_key] = _FakeAgent(time.time() - 120)

    # Pass must complete despite the wedged transport (bounded by wait_for).
    sent = await asyncio.wait_for(runner._check_session_stalls(60), timeout=5)

    # Healthy sibling was notified in the SAME pass.
    assert sent == 1
    assert len(healthy.sent) == 1
    assert healthy_key in runner._session_stall_notified
    # Wedged session: send attempted, timed out, NOT latched.
    assert wedged.send_attempts == 1
    assert wedged_key not in runner._session_stall_notified

    # Watcher ticks again: the wedged candidate is retried next pass.
    sent2 = await asyncio.wait_for(runner._check_session_stalls(60), timeout=5)
    assert sent2 == 0  # healthy already latched; wedged timed out again
    assert wedged.send_attempts == 2
    assert wedged_key not in runner._session_stall_notified
