From 5c5dd6b7ecc1b44e7e83b28ef28efb96e419badd Mon Sep 17 00:00:00 2001 From: PRATHAMESH75 Date: Sat, 11 Jul 2026 02:57:06 +0530 Subject: [PATCH] fix(agent): run cron LLM calls inline to avoid gateway deadlock (#62151) --- agent/chat_completion_helpers.py | 228 ++++++++++-------- .../agent/test_cron_inline_api_call_62151.py | 105 ++++++++ 2 files changed, 226 insertions(+), 107 deletions(-) create mode 100644 tests/agent/test_cron_inline_api_call_62151.py diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 43265df6139..546bcd42f96 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -236,73 +236,109 @@ def _check_stale_giveup(agent) -> None: ) -def should_use_direct_api_call(agent) -> bool: - """True when the non-streaming path should skip the interrupt worker thread. +def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client): + """Run one non-streaming LLM request for the active api_mode and return it. - Cron jobs run inside nested gateway thread pools (cron-scheduler → - cron-parallel → per-job pool → ``interruptible_api_call`` worker). That - extra daemon thread can wedge before HTTP on the 2nd+ API call of a - tool-using turn (#62151) while the same job succeeds via ``hermes cron - tick``. Cron has no interactive interrupt surface, so synchronous calls - on the conversation thread are safe and avoid the deadlock class. + Shared by the interrupt-worker path (``interruptible_api_call``) and the + inline path (``direct_api_call``) so the per-api_mode dispatch — codex / + anthropic / bedrock / MoA / OpenAI-compatible — lives in exactly one place. + + ``make_client(reason)`` builds the per-request OpenAI client for the codex + and OpenAI-compatible branches; the worker path uses it to register the + client with its stranger-thread abort machinery, the inline path uses it to + capture the client for its own ``finally`` close. The anthropic / bedrock / + MoA branches manage their own clients and never call it. All interrupt, + abort, cancellation, and close semantics stay in the callers — this helper + only issues the request. + """ + if agent.api_mode == "codex_responses": + request_client = make_client("codex_stream_request") + return agent._run_codex_stream( + api_kwargs, + client=request_client, + on_first_delta=getattr(agent, "_codex_on_first_delta", None), + ) + if agent.api_mode == "anthropic_messages": + return agent._anthropic_messages_create(api_kwargs) + if agent.api_mode == "bedrock_converse": + # Bedrock uses boto3 directly — no OpenAI client needed. + # normalize_converse_response produces an OpenAI-compatible + # SimpleNamespace so the rest of the agent loop can treat + # bedrock responses like chat_completions responses. + from agent.bedrock_adapter import ( + _get_bedrock_runtime_client, + invalidate_runtime_client, + is_stale_connection_error, + normalize_converse_response, + ) + region = api_kwargs.pop("__bedrock_region__", "us-east-1") + api_kwargs.pop("__bedrock_converse__", None) + client = _get_bedrock_runtime_client(region) + try: + raw_response = client.converse(**api_kwargs) + except Exception as _bedrock_exc: + # Evict the cached client on stale-connection failures + # so the outer retry loop builds a fresh client/pool. + if is_stale_connection_error(_bedrock_exc): + invalidate_runtime_client(region) + raise + return normalize_converse_response(raw_response) + if agent.provider == "moa": + # MoA is a virtual chat-completions provider backed by the + # in-process MoAClient facade. Do not rebuild a request-local + # OpenAI client from the virtual runtime metadata. + return agent.client.chat.completions.create(**api_kwargs) + request_client = make_client("chat_completion_request") + return request_client.chat.completions.create(**api_kwargs) + + +def should_use_direct_api_call(agent) -> bool: + """True when the LLM call must run inline instead of on the interrupt worker. + + ``interruptible_api_call`` / ``interruptible_streaming_api_call`` run every + request on a spawned daemon worker so the conversation loop can poll for an + interactive interrupt during the blocking HTTP round-trip. Cron jobs execute + their turn inside the gateway's *nested* thread pools (cron-scheduler → + parallel/sequential pool → per-job pool → ``run_conversation``); stacking the + interrupt worker on top of that wedges before the socket even opens on the + 2nd+ call of a tool-using turn (#62151), while the identical job runs fine + via ``hermes cron tick`` (foreground, no nested gateway pools). Cron has no + interactive interrupt surface — its only stop signal is the scheduler's + inactivity watchdog, which fires from the outer thread — so running inline + removes the deadlock class without giving anything up. This predicate is the + single extension point for any future non-interactive, nested-pool context. """ return getattr(agent, "platform", None) == "cron" -def _execute_nonstreaming_api_request(agent, api_kwargs: dict): - """Dispatch one non-streaming LLM request on the calling thread.""" - request_client = None - try: - if agent.api_mode == "codex_responses": - request_client = agent._create_request_openai_client( - reason="codex_stream_request", - api_kwargs=api_kwargs, - ) - return agent._run_codex_stream( - api_kwargs, - client=request_client, - on_first_delta=getattr(agent, "_codex_on_first_delta", None), - ) - if agent.api_mode == "anthropic_messages": - return agent._anthropic_messages_create(api_kwargs) - if agent.api_mode == "bedrock_converse": - from agent.bedrock_adapter import ( - _get_bedrock_runtime_client, - invalidate_runtime_client, - is_stale_connection_error, - normalize_converse_response, - ) - - region = api_kwargs.pop("__bedrock_region__", "us-east-1") - api_kwargs.pop("__bedrock_converse__", None) - client = _get_bedrock_runtime_client(region) - try: - raw_response = client.converse(**api_kwargs) - except Exception as bedrock_exc: - if is_stale_connection_error(bedrock_exc): - invalidate_runtime_client(region) - raise - return normalize_converse_response(raw_response) - - request_client = agent._create_request_openai_client( - reason="chat_completion_request", - api_kwargs=api_kwargs, - ) - return request_client.chat.completions.create(**api_kwargs) - finally: - if request_client is not None: - agent._close_request_openai_client( - request_client, reason="request_complete" - ) - - def direct_api_call(agent, api_kwargs: dict): - """Run a non-streaming API call synchronously on the conversation thread.""" + """Run a non-streaming LLM call inline on the conversation thread. + + Used when ``should_use_direct_api_call`` is True. Skips the interrupt worker + (whose only job is interactive-interrupt responsiveness, which this context + does not have) so the nested-pool deadlock (#62151) 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. + """ _check_stale_giveup(agent) agent._touch_activity("waiting for non-streaming API response") - response = _execute_nonstreaming_api_request(agent, api_kwargs) - _reset_stale_streak(agent) - return response + request_client_holder = {"client": None} + + def _make_client(reason: str): + client = agent._create_request_openai_client(reason=reason, api_kwargs=api_kwargs) + request_client_holder["client"] = client + return client + + try: + return _dispatch_nonstreaming_api_request( + agent, api_kwargs, make_client=_make_client + ) + finally: + if request_client_holder["client"] is not None: + agent._close_request_openai_client( + request_client_holder["client"], reason="request_complete" + ) def interruptible_api_call(agent, api_kwargs: dict): @@ -319,6 +355,12 @@ def interruptible_api_call(agent, api_kwargs: dict): the main retry loop can try again with backoff / credential rotation / provider fallback. """ + # Cron and other non-interactive, nested-pool contexts must not spawn the + # interrupt worker — it wedges before the socket opens on the 2nd+ call + # (#62151). Run inline instead. See should_use_direct_api_call. + if should_use_direct_api_call(agent): + return direct_api_call(agent, api_kwargs) + result = {"response": None, "error": None} # Cross-turn stale-call circuit breaker (#58962) — non-streaming sibling @@ -382,56 +424,19 @@ def interruptible_api_call(agent, api_kwargs: dict): def _call(): try: - if agent.api_mode == "codex_responses": - request_client = _set_request_client( + # _set_request_client registers each per-request OpenAI client with + # the stranger-thread abort machinery above; the shared dispatch + # helper builds it via this callback so the interrupt / stale-call + # detectors can force-close the worker's connection. + result["response"] = _dispatch_nonstreaming_api_request( + agent, + api_kwargs, + make_client=lambda reason: _set_request_client( agent._create_request_openai_client( - reason="codex_stream_request", - api_kwargs=api_kwargs, + reason=reason, api_kwargs=api_kwargs ) - ) - result["response"] = agent._run_codex_stream( - api_kwargs, - client=request_client, - on_first_delta=getattr(agent, "_codex_on_first_delta", None), - ) - elif agent.api_mode == "anthropic_messages": - result["response"] = agent._anthropic_messages_create(api_kwargs) - elif agent.api_mode == "bedrock_converse": - # Bedrock uses boto3 directly — no OpenAI client needed. - # normalize_converse_response produces an OpenAI-compatible - # SimpleNamespace so the rest of the agent loop can treat - # bedrock responses like chat_completions responses. - from agent.bedrock_adapter import ( - _get_bedrock_runtime_client, - invalidate_runtime_client, - is_stale_connection_error, - normalize_converse_response, - ) - region = api_kwargs.pop("__bedrock_region__", "us-east-1") - api_kwargs.pop("__bedrock_converse__", None) - client = _get_bedrock_runtime_client(region) - try: - raw_response = client.converse(**api_kwargs) - except Exception as _bedrock_exc: - # Evict the cached client on stale-connection failures - # so the outer retry loop builds a fresh client/pool. - if is_stale_connection_error(_bedrock_exc): - invalidate_runtime_client(region) - raise - result["response"] = normalize_converse_response(raw_response) - elif agent.provider == "moa": - # MoA is a virtual chat-completions provider backed by the - # in-process MoAClient facade. Do not rebuild a request-local - # OpenAI client from the virtual runtime metadata. - result["response"] = agent.client.chat.completions.create(**api_kwargs) - else: - request_client = _set_request_client( - agent._create_request_openai_client( - reason="chat_completion_request", - api_kwargs=api_kwargs, - ) - ) - result["response"] = request_client.chat.completions.create(**api_kwargs) + ), + ) except Exception as e: # If the request was cancelled by the main thread's interrupt # handler, the transport error is the expected consequence of our @@ -1942,6 +1947,15 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= if agent._interrupt_requested: raise InterruptedError("Agent interrupted before streaming API call") + # Cron and other non-interactive, nested-pool contexts deadlock on the + # spawned worker thread (#62151). They also have no stream consumer, so the + # deltas this path produces go nowhere. Delegate to the non-streaming entry + # (which runs inline via should_use_direct_api_call) exactly like the codex + # branch below — routing through the _interruptible_api_call method keeps the + # outer loop's per-request retry/refresh seam intact. + if should_use_direct_api_call(agent): + return agent._interruptible_api_call(api_kwargs) + if agent.api_mode == "codex_responses": # Codex streams internally via _run_codex_stream. The main dispatch # in _interruptible_api_call already calls it; we just need to diff --git a/tests/agent/test_cron_inline_api_call_62151.py b/tests/agent/test_cron_inline_api_call_62151.py new file mode 100644 index 00000000000..a87cbae8de4 --- /dev/null +++ b/tests/agent/test_cron_inline_api_call_62151.py @@ -0,0 +1,105 @@ +"""Regression guard for #62151 — gateway cron must not wedge on the 2nd+ call. + +Gateway-fired cron jobs hung forever on the 2nd+ API call because both the +non-streaming (``interruptible_api_call``) and the default streaming +(``interruptible_streaming_api_call``) paths run the request on a spawned +daemon worker thread. Inside the gateway's nested cron thread pools that extra +worker wedged before the socket opened; the same job succeeded via ``hermes +cron tick`` (foreground, no nested pools). Cron has no interactive interrupt +surface, so both paths now run inline on the conversation thread for the +``cron`` platform. + +These tests pin: (1) the inline gate is cron-only, (2) the inline call runs on +the *calling* thread — no worker is spawned — for both entry points, and (3) +the shared dispatch closes the per-request client. +""" + +import threading +from types import SimpleNamespace +from unittest.mock import MagicMock + +from agent.chat_completion_helpers import ( + direct_api_call, + interruptible_api_call, + interruptible_streaming_api_call, + should_use_direct_api_call, +) + + +def _make_agent(*, 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._close_request_openai_client = MagicMock() + return agent + + +def test_should_use_direct_api_call_only_for_cron_platform(): + assert should_use_direct_api_call(_make_agent(platform="cron")) is True + assert should_use_direct_api_call(_make_agent(platform="cli")) is False + assert should_use_direct_api_call(_make_agent(platform="telegram")) is False + assert should_use_direct_api_call(_make_agent(platform=None)) is False + + +def test_direct_api_call_runs_inline_and_closes_client(): + agent = _make_agent() + caller_tid = threading.get_ident() + ran_on = {} + fake_client = MagicMock() + + def _create(**_kwargs): + ran_on["tid"] = threading.get_ident() + return fake_client + + fake_client.chat.completions.create.return_value = SimpleNamespace(id="resp") + agent._create_request_openai_client.side_effect = _create + + resp = direct_api_call(agent, {"model": "m", "messages": []}) + + assert resp.id == "resp" + # Inline: the request ran on the calling thread, no worker was spawned. + assert ran_on["tid"] == caller_tid + assert agent._close_request_openai_client.call_count == 1 + + +def test_interruptible_api_call_routes_cron_inline_no_worker_thread(): + agent = _make_agent() + caller_tid = threading.get_ident() + fake_client = MagicMock() + ran_on = {} + + def _create(**_kwargs): + ran_on["tid"] = threading.get_ident() + return fake_client + + fake_client.chat.completions.create.return_value = SimpleNamespace(id="first") + agent._create_request_openai_client.side_effect = _create + + resp = interruptible_api_call(agent, {"model": "m", "messages": []}) + + assert resp.id == "first" + assert ran_on["tid"] == caller_tid # no daemon worker thread + + +def test_interruptible_streaming_api_call_routes_cron_via_nonstream_method(): + """Streaming is the default even for cron — the gate must catch it too. + + It delegates to the ``_interruptible_api_call`` method (which itself runs + inline for cron) rather than calling ``direct_api_call`` directly, so the + outer loop's per-request retry/refresh seam — which patches that method — + stays intact (regression from the codex 401-refresh path). + """ + agent = _make_agent() + sentinel = SimpleNamespace(id="via-nonstream") + agent._interruptible_api_call = MagicMock(return_value=sentinel) + + resp = interruptible_streaming_api_call( + agent, {"model": "m", "messages": []}, on_first_delta=lambda: None + ) + + assert resp is sentinel + agent._interruptible_api_call.assert_called_once()