"""Regression guard for #80759 — the inline non-streaming call must be bounded.

Cron turns (and delegated children) are routed by ``should_use_direct_api_call``
onto ``direct_api_call``, which ran the request inline with no stale detector:
the abort plumbing was registered but nothing ever invoked it. A provider that
accepted the request and then went silent — connection held open, zero bytes,
no error — hung the cron run until an external actor killed it (which also
orphaned the execution row). The only other bound was the 1800s-default httpx
read timeout, and the job-level inactivity monitor was observed not to fire.

These tests pin the watchdog contract: it aborts the in-flight sockets through
the already-registered abort hook, surfaces a retryable ``TimeoutError`` (never
``InterruptedError``), feeds the cross-turn stale circuit breaker, and stays
out of the way of a healthy call. They also pin #85252: the keepalive httpx
client uses ``read=None``, so a stranger-thread abort that finds no sockets
must not leave the call unbounded — ``direct_api_call`` injects a per-call
read timeout matching the stale budget as a hard backstop.
"""

import sys
import threading
import time
import types
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

sys.modules.setdefault("fire", types.SimpleNamespace(Fire=lambda *a, **k: None))
sys.modules.setdefault("firecrawl", types.SimpleNamespace(Firecrawl=object))
sys.modules.setdefault("fal_client", types.SimpleNamespace())

import run_agent

import agent.chat_completion_helpers as chat_completion_helpers
from agent.chat_completion_helpers import direct_api_call


def _make_agent(*, stale_timeout, platform="cron"):
    agent = MagicMock()
    agent.platform = platform
    agent.api_mode = "chat_completions"
    agent.provider = "openrouter"
    agent._interrupt_requested = False
    agent._consecutive_stale_streams = 0
    agent._touch_activity = MagicMock()
    agent._buffer_status = MagicMock()
    agent._create_request_openai_client = MagicMock()
    agent._close_request_openai_client = MagicMock()
    agent._abort_request_openai_client = MagicMock()
    agent._compute_non_stream_stale_timeout = lambda api_payload: stale_timeout
    return agent


def _stalling_client(agent, *, aborted):
    """A client whose request blocks until the watchdog aborts its sockets."""
    fake_client = MagicMock()
    release_after_abort = threading.Event()

    def _abort(client, reason):
        aborted.append(reason)
        release_after_abort.set()

    def _stalled_request(**_kwargs):
        # The provider accepted the request and went silent. The socket
        # shutdown is what unblocks it, exactly as in production.
        release_after_abort.wait()
        raise ConnectionError("socket shut down")

    fake_client.chat.completions.create.side_effect = _stalled_request
    agent._abort_request_openai_client.side_effect = _abort
    agent._create_request_openai_client.return_value = fake_client
    return fake_client


def test_stalled_inline_call_is_aborted_and_raises_retryable_timeout():
    agent = _make_agent(stale_timeout=0.2)
    aborted: list[str] = []
    _stalling_client(agent, aborted=aborted)

    with pytest.raises(TimeoutError) as excinfo:
        direct_api_call(agent, {"model": "m", "messages": []})

    assert aborted == ["stale_call_kill"]
    assert "no response" in str(excinfo.value)


def test_inline_cron_openai_codex_keeps_large_context_stale_floor(monkeypatch):
    """#69734: cron Codex runs inline, so the inline stale budget must keep the
    openai-codex large-context floor the worker path applied — else a healthy
    >10k-token cron turn is killed at the 90s default."""
    from agent.chat_completion_helpers import _resolve_direct_stale_timeout, should_use_direct_api_call

    for key in ("HERMES_API_CALL_STALE_TIMEOUT", "HERMES_STREAM_STALE_TIMEOUT", "HERMES_CODEX_HARD_TIMEOUT_SECONDS"):
        monkeypatch.delenv(key, raising=False)
    agent = run_agent.AIAgent(
        model="gpt-5.5", provider="openai-codex", api_mode="codex_responses",
        base_url="https://chatgpt.com/backend-api/codex", api_key="x", quiet_mode=True,
        skip_context_files=True, skip_memory=True, platform="cron",
    )
    api_kwargs = {"model": "gpt-5.5", "input": [{"role": "user", "content": "x " * 60000}]}
    assert should_use_direct_api_call(agent) is True
    assert _resolve_direct_stale_timeout(agent, api_kwargs) >= 600.0


def test_inline_local_responses_endpoint_keeps_its_configured_stale_budget(monkeypatch):
    """The hosted Codex floor/hard cap must not clamp a local Responses server's
    configured stale budget on the inline path either (same rule as the worker path)."""
    from agent.chat_completion_helpers import _resolve_direct_stale_timeout

    monkeypatch.delenv("HERMES_CODEX_HARD_TIMEOUT_SECONDS", raising=False)
    monkeypatch.setenv("HERMES_API_CALL_STALE_TIMEOUT", "3000")
    agent = run_agent.AIAgent(
        model="local-model", provider="custom", api_mode="codex_responses",
        base_url="http://127.0.0.1:8080/v1", api_key="x", quiet_mode=True,
        skip_context_files=True, skip_memory=True, platform="cron",
    )
    api_kwargs = {"model": "local-model", "input": [{"role": "user", "content": "x " * 60000}]}
    assert _resolve_direct_stale_timeout(agent, api_kwargs) == 3000.0


def test_watchdog_kill_feeds_the_cross_turn_stale_circuit_breaker():
    """Without a bump the #58962 breaker can never trip for cron sessions."""
    agent = _make_agent(stale_timeout=0.2)
    _stalling_client(agent, aborted=[])

    with pytest.raises(TimeoutError):
        direct_api_call(agent, {"model": "m", "messages": []})

    assert agent._consecutive_stale_streams == 1


def test_retry_after_a_watchdog_kill_gets_a_fresh_pool_and_succeeds():
    agent = _make_agent(stale_timeout=0.2)
    _stalling_client(agent, aborted=[])

    with pytest.raises(TimeoutError):
        direct_api_call(agent, {"model": "m", "messages": []})

    # The kill must really close the wire client so the retry rebuilds it.
    assert agent._close_request_openai_client.call_args.kwargs["reason"] == (
        "request_error_cleanup"
    )

    healthy_client = MagicMock()
    healthy_client.chat.completions.create.return_value = SimpleNamespace(id="ok")
    agent._create_request_openai_client.side_effect = None
    agent._create_request_openai_client.return_value = healthy_client

    assert direct_api_call(agent, {"model": "m", "messages": []}).id == "ok"
    assert agent._consecutive_stale_streams == 0


def test_healthy_call_is_untouched_by_the_watchdog():
    agent = _make_agent(stale_timeout=30.0)
    fake_client = MagicMock()
    fake_client.chat.completions.create.return_value = SimpleNamespace(id="fast")
    agent._create_request_openai_client.return_value = fake_client

    assert direct_api_call(agent, {"model": "m", "messages": []}).id == "fast"
    agent._abort_request_openai_client.assert_not_called()
    assert agent._close_request_openai_client.call_args.kwargs["reason"] == (
        "request_complete"
    )


def test_local_endpoint_infinite_budget_leaves_the_watchdog_disarmed():
    """``_compute_non_stream_stale_timeout`` returns inf for a local endpoint
    on the implicit default — that opt-out must survive on this path too."""
    agent = _make_agent(stale_timeout=float("inf"))
    fake_client = MagicMock()
    started = threading.Event()
    release = threading.Event()

    def _slow(**_kwargs):
        started.set()
        assert release.wait(timeout=2.0)
        return SimpleNamespace(id="slow-but-healthy")

    fake_client.chat.completions.create.side_effect = _slow
    agent._create_request_openai_client.return_value = fake_client

    box = {}

    def _run():
        box["response"] = direct_api_call(agent, {"model": "m", "messages": []})

    worker = threading.Thread(target=_run, daemon=True)
    worker.start()
    assert started.wait(timeout=2.0)
    time.sleep(0.3)
    release.set()
    worker.join(timeout=3.0)

    assert box["response"].id == "slow-but-healthy"
    agent._abort_request_openai_client.assert_not_called()




# ---------------------------------------------------------------------------
# End-to-end: a real AIAgent on the real cron routing + client lifecycle.
# ---------------------------------------------------------------------------


class _StallingWireClient:
    """An OpenAI-shaped client whose request stalls until its sockets die."""

    def __init__(self):
        self._client = SimpleNamespace(is_closed=False)
        self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create))
        self.responses = SimpleNamespace()
        self.close_calls = 0
        self.request_started = threading.Event()
        self.sockets_shut_down = threading.Event()

    def _create(self, **_kwargs):
        self.request_started.set()
        self.sockets_shut_down.wait()
        raise ConnectionError("socket shut down")

    def close(self):
        self.close_calls += 1
        self._client.is_closed = True


def _build_cron_agent(monkeypatch):
    agent = run_agent.AIAgent.__new__(run_agent.AIAgent)
    agent.platform = "cron"
    agent.api_mode = "chat_completions"
    agent.provider = "openrouter"
    agent.base_url = "https://openrouter.ai/api/v1"
    agent._base_url = agent.base_url
    agent.model = "some/model"
    agent.log_prefix = ""
    agent.quiet_mode = True
    agent._interrupt_requested = False
    agent._interrupt_message = None
    agent._client_lock = threading.RLock()
    agent._client_kwargs = {"api_key": "***", "base_url": agent.base_url}
    agent.stream_delta_callback = None
    agent._stream_callback = None
    agent.reasoning_callback = None
    agent.status_callback = None
    monkeypatch.setenv("HERMES_API_CALL_STALE_TIMEOUT", "0.3")
    return agent


def test_e2e_cron_turn_is_bounded_through_the_real_agent_routing(monkeypatch):
    """The real chain: cron platform → ``should_use_direct_api_call`` →
    ``direct_api_call`` → the agent's own stale-timeout resolver → the real
    cross-thread abort → a retryable ``TimeoutError`` on the caller."""
    wire = _StallingWireClient()
    agent = _build_cron_agent(monkeypatch)
    agent.client = wire
    timer_intervals = []

    class RequestStartedTimer:
        """Run the real watchdog callback after dispatch reaches the wire.

        CI process starvation can delay either side of a sub-second Timer by
        tens of seconds. The event fixes their causal order without replacing
        the routing, timeout resolver, watchdog callback, or abort lifecycle.
        """

        def __init__(self, interval, function):
            timer_intervals.append(interval)
            self._function = function
            self._cancelled = threading.Event()
            self.name = None
            self.daemon = False

        def _run(self):
            wire.request_started.wait()
            if not self._cancelled.is_set():
                self._function()

        def start(self):
            threading.Thread(
                target=self._run, name=self.name, daemon=self.daemon
            ).start()

        def cancel(self):
            self._cancelled.set()

    threading_proxy = SimpleNamespace(**vars(threading))
    threading_proxy.Timer = RequestStartedTimer
    monkeypatch.setattr(chat_completion_helpers, "threading", threading_proxy)
    monkeypatch.setattr("agent.process_bootstrap.OpenAI", lambda **_kwargs: wire)
    monkeypatch.setattr(
        run_agent.AIAgent,
        "_force_close_tcp_sockets",
        lambda self, client: (client.sockets_shut_down.set(), 1)[1],
    )

    with pytest.raises(TimeoutError):
        agent._interruptible_api_call({"model": agent.model, "messages": []})

    assert timer_intervals == [pytest.approx(0.3)]
    assert wire.request_started.is_set()
    assert wire.sockets_shut_down.is_set()
    # The aborted pool is poisoned, so the killed client is really closed
    # instead of being cached for the retry.
    assert wire.close_calls == 1


# ---------------------------------------------------------------------------
# Salvage follow-up (#75301 state discipline): locked lifecycle transitions.
# ---------------------------------------------------------------------------


def test_interrupt_abort_is_not_misclassified_as_provider_staleness():
    """A user/monitor interrupt that kills the call must not advance the
    cross-turn stale circuit breaker, even if the stale timer fires right
    after — the ``cancelled`` flag owns the outcome (#75301 design)."""
    agent = _make_agent(stale_timeout=30.0)
    release_after_abort = threading.Event()
    fake_client = MagicMock()

    def _stalled_request(**_kwargs):
        assert release_after_abort.wait(timeout=5.0)
        raise ConnectionError("socket shut down")

    fake_client.chat.completions.create.side_effect = _stalled_request
    agent._create_request_openai_client.return_value = fake_client

    def _abort(client, reason):
        release_after_abort.set()

    agent._abort_request_openai_client.side_effect = _abort

    interrupt_fired = threading.Event()

    def _interrupt_soon():
        # Wait until the abort hook is registered, then interrupt like
        # run_agent.interrupt() does (registered under the same name).
        deadline = time.time() + 2.0
        while time.time() < deadline:
            hook = agent._active_request_abort
            if callable(hook) and not isinstance(hook, MagicMock):
                agent._interrupt_requested = True
                # interrupt owns the outcome...
                assert hook("interrupt_abort") is False
                # ...so a stale timer racing in afterwards is inert:
                assert hook("stale_call_kill") is False
                interrupt_fired.set()
                return
            time.sleep(0.005)

    worker = threading.Thread(target=_interrupt_soon, daemon=True)
    worker.start()

    with pytest.raises(InterruptedError):
        direct_api_call(agent, {"model": "m", "messages": []})

    assert interrupt_fired.wait(timeout=1.0)
    assert agent._consecutive_stale_streams == 0, (
        "interrupt was misclassified as provider staleness"
    )


def test_late_stale_timer_after_completion_is_inert():
    """A timer callback that loses the race to a completed request must not
    bump the streak: ``done`` is set under the lock before the unwind."""
    agent = _make_agent(stale_timeout=30.0)
    fake_client = MagicMock()
    fake_client.chat.completions.create.return_value = SimpleNamespace(id="ok")
    agent._create_request_openai_client.return_value = fake_client

    captured = {}
    real_timer = threading.Timer

    class CapturingTimer(real_timer):
        def __init__(self, interval, function, *a, **k):
            captured["fn"] = function
            super().__init__(interval, function, *a, **k)

    with pytest.MonkeyPatch.context() as mp:
        mp.setattr(threading, "Timer", CapturingTimer)
        assert direct_api_call(agent, {"model": "m", "messages": []}).id == "ok"

    # Fire the (already-cancelled) timer callback manually, simulating a
    # timer thread that had already dequeued before cancel().
    captured["fn"]()
    assert agent._consecutive_stale_streams == 0
    agent._abort_request_openai_client.assert_not_called()


def test_timer_firing_before_client_registration_fails_the_dispatch():
    """Registration race: if the budget expires while the client is still
    being constructed, the freshly-registered client is aborted and the call
    fails with a retryable TimeoutError instead of opening a new socket
    after the only watchdog was spent."""
    agent = _make_agent(stale_timeout=0.05)
    fake_client = MagicMock()
    fake_client.chat.completions.create.return_value = SimpleNamespace(id="late")

    def _slow_create(*, reason, api_kwargs):
        # The timer (50ms budget) fires while construction is in flight.
        time.sleep(0.4)
        return fake_client

    agent._create_request_openai_client.side_effect = _slow_create

    with pytest.raises(TimeoutError):
        direct_api_call(agent, {"model": "m", "messages": []})

    fake_client.chat.completions.create.assert_not_called()
    agent._abort_request_openai_client.assert_called_once_with(
        fake_client, reason="stale_call_kill"
    )


def test_resolver_exception_propagates_instead_of_disarming_the_watchdog():
    """A raising stale-timeout resolver must propagate (fail closed), not be
    swallowed into an infinite budget that silently reinstates the hang."""
    agent = _make_agent(stale_timeout=0.2)

    def _broken_resolver(api_payload):
        raise RuntimeError("resolver regression")

    agent._compute_non_stream_stale_timeout = _broken_resolver

    with pytest.raises(RuntimeError, match="resolver regression"):
        direct_api_call(agent, {"model": "m", "messages": []})


# ---------------------------------------------------------------------------
# #85252: hard socket bound when stranger-thread abort cannot kill the recv.
# ---------------------------------------------------------------------------






def test_inline_call_passes_hard_read_timeout_to_the_sdk():
    """The bound has to actually reach chat.completions.create — a helper
    that is never wired in would leave cron on read=None (#85252)."""
    agent = _make_agent(stale_timeout=0.5)
    fake_client = MagicMock()
    captured = {}

    def _create(**kwargs):
        captured["timeout"] = kwargs.get("timeout")
        return SimpleNamespace(id="ok")

    fake_client.chat.completions.create.side_effect = _create
    agent._create_request_openai_client.return_value = fake_client

    assert direct_api_call(agent, {"model": "m", "messages": []}).id == "ok"
    timeout = captured["timeout"]
    assert timeout is not None
    assert timeout.read == 0.5


def test_inline_call_does_not_override_explicit_timeout():
    """A transport/provider that already set timeout= must keep it."""
    agent = _make_agent(stale_timeout=30.0)
    fake_client = MagicMock()
    captured = {}

    def _create(**kwargs):
        captured["timeout"] = kwargs.get("timeout")
        return SimpleNamespace(id="ok")

    fake_client.chat.completions.create.side_effect = _create
    agent._create_request_openai_client.return_value = fake_client

    assert direct_api_call(
        agent, {"model": "m", "messages": [], "timeout": 12.0}
    ).id == "ok"
    assert captured["timeout"] == 12.0


def test_infinite_budget_does_not_inject_a_hard_timeout():
    agent = _make_agent(stale_timeout=float("inf"))
    fake_client = MagicMock()
    captured = {}

    def _create(**kwargs):
        captured["timeout"] = kwargs.get("timeout")
        return SimpleNamespace(id="ok")

    fake_client.chat.completions.create.side_effect = _create
    agent._create_request_openai_client.return_value = fake_client

    assert direct_api_call(agent, {"model": "m", "messages": []}).id == "ok"
    assert "timeout" not in captured or captured["timeout"] is None