mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-15 14:22:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(agent): run cron LLM calls inline to avoid gateway deadlock (#62151)

Browse source
This commit is contained in:
PRATHAMESH75 2026-07-11 02:57:06 +05:30 • committed by kshitij
parent 1234f39e31
commit 5c5dd6b7ec
2 changed files with 226 additions and 107 deletions
Download patch file Download diff file Expand all files Collapse all files

228
agent/chat_completion_helpers.py
View file

@ -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

105
tests/agent/test_cron_inline_api_call_62151.py Normal file
View file

@ -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()