"""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. """ 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 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, release_after=5.0): """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. if not release_after_abort.wait(timeout=release_after): raise AssertionError("watchdog never aborted the stalled request") 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) started = time.time() with pytest.raises(TimeoutError) as excinfo: direct_api_call(agent, {"model": "m", "messages": []}) elapsed = time.time() - started assert aborted == ["stale_call_kill"] assert "no response" in str(excinfo.value) assert elapsed < 4.0, "watchdog did not bound the call" def test_watchdog_abort_never_surfaces_as_interrupted_error(): """InterruptedError means "the user wants to stop" — the outer loop does not retry it. A watchdog abort must stay retryable.""" agent = _make_agent(stale_timeout=0.2) _stalling_client(agent, aborted=[]) with pytest.raises(TimeoutError): direct_api_call(agent, {"model": "m", "messages": []}) 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() def test_watchdog_uses_the_same_budget_as_the_interrupt_worker_path(): """The budget comes from ``_compute_non_stream_stale_timeout`` — the same resolver the worker path's stale detector uses — with the live request payload, so provider config and context scaling both apply.""" seen: list[dict] = [] agent = _make_agent(stale_timeout=30.0) agent._compute_non_stream_stale_timeout = lambda payload: ( seen.append(payload) or 30.0 ) fake_client = MagicMock() fake_client.chat.completions.create.return_value = SimpleNamespace(id="ok") agent._create_request_openai_client.return_value = fake_client payload = {"model": "m", "messages": [{"role": "user", "content": "hi"}]} direct_api_call(agent, payload) assert seen == [payload] # --------------------------------------------------------------------------- # 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.sockets_shut_down = threading.Event() def _create(self, **_kwargs): if not self.sockets_shut_down.wait(timeout=5.0): raise AssertionError("watchdog never shut the stalled request down") 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 monkeypatch.setattr(run_agent, "OpenAI", lambda **_kwargs: wire) monkeypatch.setattr( run_agent.AIAgent, "_force_close_tcp_sockets", lambda self, client: (client.sockets_shut_down.set(), 1)[1], ) started = time.time() with pytest.raises(TimeoutError): agent._interruptible_api_call({"model": agent.model, "messages": []}) elapsed = time.time() - started assert elapsed < 4.0, "cron turn was not bounded by the watchdog" # 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": []})