fix(cron): bound the inline non-streaming call with a stale watchdog
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
This commit is contained in:
HexLab98
@@ -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:
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user