From cb066a971bbfdeb64f95a31cf59a49a56007e1da Mon Sep 17 00:00:00 2001
From: HexLab98 <liruixinch@outlook.com>
Date: Fri, 7 Aug 2026 12:05:48 +0700
Subject: [PATCH] fix(cron): bound the inline non-streaming call with a stale
 watchdog
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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

Arm a watchdog timer on the same budget the interrupt worker's poll loop
uses, so these turns get exactly the patience every other non-streaming
request already gets. On expiry it only aborts the in-flight sockets
through the already-registered hook — it never issues a request, so the
inline / no-worker property that fixes the nested-pool deadlock is
preserved — bumps the cross-turn stale circuit breaker, and surfaces a
retryable TimeoutError so the outer loop reconnects on a fresh pool.

Fixes #80759
---
 agent/chat_completion_helpers.py              | 100 ++++++-
 .../test_cron_direct_api_call_watchdog.py     | 263 ++++++++++++++++++
 2 files changed, 359 insertions(+), 4 deletions(-)
 create mode 100644 tests/cron/test_cron_direct_api_call_watchdog.py

diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 56fc1dac71..66f0178688 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -564,28 +564,62 @@ def should_use_direct_api_call(agent) -> bool:
 _DIRECT_API_ACTIVITY_HEARTBEAT_SECONDS = 15.0
 
 
+def _resolve_direct_stale_timeout(agent, api_kwargs: dict) -> float:
+    """Stale budget for the inline non-streaming call.
+
+    Same derivation the interrupt-worker path uses for its stale-call
+    detector (provider ``stale_timeout_seconds`` →
+    ``HERMES_API_CALL_STALE_TIMEOUT`` → reasoning-model floor → context-size
+    scaling, ``inf`` for a local endpoint on the implicit default), so cron and
+    delegated turns get exactly the patience every other non-streaming request
+    already gets.
+
+    A non-numeric result — an agent stub that never implements the resolver —
+    leaves the watchdog disarmed rather than arming it on a bogus budget.
+    """
+    resolver = getattr(agent, "_compute_non_stream_stale_timeout", None)
+    if not callable(resolver):
+        return float("inf")
+    try:
+        value = resolver(api_kwargs)
+    except Exception:
+        logger.debug("non-stream stale timeout resolution failed", exc_info=True)
+        return float("inf")
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return float("inf")
+    return float(value)
+
+
 def direct_api_call(agent, api_kwargs: dict):
     """Run a non-streaming LLM call inline on the conversation thread.
 
     Used when ``should_use_direct_api_call`` is True (cron turns and
     delegated children). Skips the interrupt worker (whose only job is
     interactive-interrupt responsiveness, which these contexts do not have)
-    so the nested-pool deadlock (#62151, #60203) cannot occur. Because the
-    request runs in-flight normally, the per-request OpenAI client's own httpx
-    timeout (provider ``request_timeout_seconds`` / ``HERMES_API_TIMEOUT``) bounds
-    a genuinely hung provider — the same bound interactive calls already rely on.
+    so the nested-pool deadlock (#62151, #60203) cannot occur.
 
     While the inline request blocks, a lightweight activity heartbeat keeps
     ``last_activity_ts`` advancing. Subagents use this path (non-streaming),
     and without mid-call ticks the async stall monitor / sync heartbeat treat
     a slow-but-healthy local model wait as "no progress" and interrupt around
     450s — surfacing as ``Operation interrupted: waiting for model response``.
+
+    A stale-call watchdog bounds the request the same way the interrupt
+    worker's poll loop does (#80759). The httpx read timeout alone is not a
+    usable bound: it defaults to 1800s and a provider that accepts the request
+    and then goes silent (connection held open, zero bytes, no error) never
+    trips it, so a cron run hangs until something external kills it — which
+    also orphans the execution row. The watchdog aborts the in-flight sockets
+    through the already-registered abort hook and surfaces a retryable
+    ``TimeoutError`` so the outer retry loop reconnects with backoff /
+    credential rotation / provider fallback.
     """
     _check_stale_giveup(agent)
     agent._touch_activity("waiting for non-streaming API response")
     request_client_holder = {"client": None}
     request_client_lock = threading.Lock()
     activity_hb_stop = threading.Event()
+    stale_fired = threading.Event()
 
     def _abort_active_request(reason: str) -> None:
         """Abort the inline request from a watchdog/interrupt thread."""
@@ -628,6 +662,52 @@ def direct_api_call(agent, api_kwargs: dict):
     )
     activity_hb.start()
 
+    call_start = time.time()
+    stale_timeout = _resolve_direct_stale_timeout(agent, api_kwargs)
+
+    def _on_stale() -> None:
+        # Runs on the timer thread. It only aborts the in-flight sockets —
+        # it never issues a request — so the inline / no-worker property that
+        # fixes #62151 and #60203 is preserved.
+        if stale_fired.is_set():
+            return
+        stale_fired.set()
+        elapsed = time.time() - call_start
+        model = api_kwargs.get("model", "unknown")
+        logger.warning(
+            "Inline non-streaming API call stale for %.0fs (threshold %.0fs). "
+            "model=%s context=~%s tokens. Killing connection.",
+            elapsed,
+            stale_timeout,
+            model,
+            f"{estimate_request_context_tokens(api_kwargs):,}",
+        )
+        try:
+            agent._buffer_status(
+                f"⚠️ No response from provider for {int(elapsed)}s "
+                f"(non-streaming, model: {model}). Aborting call."
+            )
+        except Exception:
+            logger.debug("stale status buffering failed", exc_info=True)
+        _bump_stale_streak(agent)
+        try:
+            agent._touch_activity(
+                f"stale non-streaming call killed after {int(elapsed)}s"
+            )
+        except Exception:
+            logger.debug("stale activity touch failed", exc_info=True)
+        try:
+            _abort_active_request("stale_call_kill")
+        except Exception:
+            logger.debug("stale abort failed", exc_info=True)
+
+    stale_watchdog = None
+    if math.isfinite(stale_timeout) and stale_timeout > 0:
+        stale_watchdog = threading.Timer(stale_timeout, _on_stale)
+        stale_watchdog.name = "direct-api-stale-watchdog"
+        stale_watchdog.daemon = True
+        stale_watchdog.start()
+
     # Only a clean return may report the reuse reason (request_complete):
     # after an error or interrupt the wire client is really closed so the
     # retry builds a fresh pool (see _REQUEST_CLIENT_REUSE_REASONS).
@@ -639,6 +719,16 @@ def direct_api_call(agent, api_kwargs: dict):
     except Exception:
         if getattr(agent, "_interrupt_requested", False):
             raise InterruptedError("Agent interrupted during API call") from None
+        if stale_fired.is_set():
+            # The transport error is the expected consequence of our own
+            # abort. Raise a retryable TimeoutError (never InterruptedError,
+            # which the outer loop treats as "the user wants to stop") so the
+            # retry loop reconnects on a fresh pool.
+            raise TimeoutError(
+                f"Non-streaming API call timed out after "
+                f"{int(time.time() - call_start)}s with no response "
+                f"(threshold: {int(stale_timeout)}s)"
+            ) from None
         raise
     else:
         if getattr(agent, "_interrupt_requested", False):
@@ -647,6 +737,8 @@ def direct_api_call(agent, api_kwargs: dict):
         succeeded = True
         return response
     finally:
+        if stale_watchdog is not None:
+            stale_watchdog.cancel()
         activity_hb_stop.set()
         activity_hb.join(timeout=2.0)
         if getattr(agent, "_active_request_abort", None) is _abort_active_request:
diff --git a/tests/cron/test_cron_direct_api_call_watchdog.py b/tests/cron/test_cron_direct_api_call_watchdog.py
new file mode 100644
index 0000000000..8640e2fd6f
--- /dev/null
+++ b/tests/cron/test_cron_direct_api_call_watchdog.py
@@ -0,0 +1,263 @@
+"""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