mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-21 16:18:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #67238 from NousResearch/bb/anthropic-owner-thread-abort

Browse source 
fix(agent): request-local Anthropic clients so the stale/interrupt watchdog never corrupts SQLite (#67142, supersedes #51688)
This commit is contained in:
brooklyn! 2026-07-18 22:37:40 -04:00 • committed by GitHub
commit e53f87fe69
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 423 additions and 103 deletions
Download patch file Download diff file Expand all files Collapse all files

168
agent/chat_completion_helpers.py
View file

@ -372,13 +372,14 @@ def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client):
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.
``make_client(reason, kind=...)`` builds the per-request client for the
codex / OpenAI-compatible (``kind="openai"``) and anthropic
(``kind="anthropic_messages"``) 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
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")
@ -388,7 +389,13 @@ def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client):
on_first_delta=getattr(agent, "_codex_on_first_delta", None),
)
if agent.api_mode == "anthropic_messages":
return agent._anthropic_messages_create(api_kwargs)
# #67142: use a request-local Anthropic client so the stale/interrupt
# watchdog aborts sockets from the stranger thread while the worker
# owns the SDK close — never closing the shared client mid-flight.
request_client = make_client(
"anthropic_messages_request", kind="anthropic_messages"
)
return agent._anthropic_messages_create(api_kwargs, client=request_client)
if agent.api_mode == "bedrock_converse":
# Bedrock uses boto3 directly — no OpenAI client needed.
# normalize_converse_response produces an OpenAI-compatible
@ -458,7 +465,11 @@ def direct_api_call(agent, api_kwargs: dict):
if request_client is not None:
agent._abort_request_openai_client(request_client, reason=reason)
def _make_client(reason: str):
def _make_client(reason: str, kind: str = "openai"):
# direct_api_call only runs for OpenAI-wire chat_completions cron
# requests (see should_use_direct_api_call), so the anthropic branch of
# the dispatch — the only caller that passes kind — is never reached
# here; the ``kind`` parameter exists purely for signature parity.
client = agent._create_request_openai_client(reason=reason, api_kwargs=api_kwargs)
with request_client_lock:
request_client_holder["client"] = client
@ -517,6 +528,10 @@ def interruptible_api_call(agent, api_kwargs: dict):
_check_stale_giveup(agent)
request_client_holder = {"client": None, "owner_tid": None}
# Transport kind of the registered request client ("openai" or
# "anthropic_messages") so _close_request_client_once routes to the right
# abort/close helpers (#67142).
request_client_kind = {"value": "openai"}
request_client_lock = threading.Lock()
# Request-local cancellation flag. Distinct from agent._interrupt_requested
# because that flag is cleared at run_conversation() turn boundaries, but
@ -528,9 +543,10 @@ def interruptible_api_call(agent, api_kwargs: dict):
# hang.)
_request_cancelled = {"value": False}
def _set_request_client(client):
def _set_request_client(client, *, kind: str = "openai"):
with request_client_lock:
request_client_holder["client"] = client
request_client_kind["value"] = kind
# #29507: stamp the owning thread so a stranger-thread interrupt
# only shuts the connection down rather than racing the worker
# for FD ownership during ``client.close()``.
@ -564,24 +580,34 @@ def interruptible_api_call(agent, api_kwargs: dict):
request_client_holder["owner_tid"] = None
if request_client is None:
return
if stranger_thread:
kind = request_client_kind.get("value", "openai")
if kind == "anthropic_messages":
if stranger_thread:
agent._abort_request_anthropic_client(request_client, reason=reason)
else:
agent._close_request_anthropic_client(request_client, reason=reason)
elif stranger_thread:
agent._abort_request_openai_client(request_client, reason=reason)
else:
agent._close_request_openai_client(request_client, reason=reason)
def _call():
try:
# _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.
# _set_request_client registers each per-request client with the
# stranger-thread abort machinery above; the shared dispatch helper
# builds it via this callback (openai- or anthropic-kind) so the
# interrupt / stale-call detectors can force-close the worker's
# connection without touching the shared client (#67142).
result["response"] = _dispatch_nonstreaming_api_request(
agent,
api_kwargs,
make_client=lambda reason: _set_request_client(
agent._create_request_openai_client(
make_client=lambda reason, kind="openai": _set_request_client(
agent._create_request_anthropic_client(reason=reason)
if kind == "anthropic_messages"
else agent._create_request_openai_client(
reason=reason, api_kwargs=api_kwargs
)
),
kind=kind,
),
)
except Exception as e:
@ -893,11 +919,10 @@ def interruptible_api_call(agent, api_kwargs: dict):
f"Aborting call."
)
try:
if agent.api_mode == "anthropic_messages":
agent._anthropic_client.close()
agent._rebuild_anthropic_client()
else:
_close_request_client_once("stale_call_kill")
# #67142: routes by client kind — anthropic now aborts the
# request-local client's sockets from this poll (stranger)
# thread instead of closing the shared _anthropic_client.
_close_request_client_once("stale_call_kill")
except Exception:
pass
# Circuit breaker (#58962): count the stale kill. See the
@ -933,13 +958,12 @@ def interruptible_api_call(agent, api_kwargs: dict):
)
# Force-close the in-flight worker-local HTTP connection to stop
# token generation without poisoning the shared client used to
# seed future retries.
# seed future retries. #67142: for anthropic this aborts the
# request-local client's sockets from this poll (stranger) thread
# rather than closing the shared _anthropic_client, which could
# release a TLS FD mid-SSL-BIO and corrupt an unrelated SQLite DB.
try:
if agent.api_mode == "anthropic_messages":
agent._anthropic_client.close()
agent._rebuild_anthropic_client()
else:
_close_request_client_once("interrupt_abort")
_close_request_client_once("interrupt_abort")
except Exception:
pass
raise InterruptedError("Agent interrupted during API call")
@ -2393,6 +2417,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
_check_stale_giveup(agent)
request_client_holder = {"client": None, "diag": None, "owner_tid": None}
# Transport kind of the registered request client — see the non-streaming
# variant. Routes _close_request_client_once to anthropic vs openai abort/
# close helpers (#67142).
request_client_kind = {"value": "openai"}
request_client_lock = threading.Lock()
# Request-local cancellation flag — see interruptible_api_call for the full
# rationale. The streaming retry loop is where the 7-minute cascading-
@ -2403,9 +2431,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
# exit immediately instead of retrying. (PR #6600.)
_request_cancelled = {"value": False}
def _set_request_client(client):
def _set_request_client(client, *, kind: str = "openai"):
with request_client_lock:
request_client_holder["client"] = client
request_client_kind["value"] = kind
# See #29507 explanation in the non-streaming variant above.
request_client_holder["owner_tid"] = threading.get_ident()
return client
@ -2428,7 +2457,13 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
request_client_holder["owner_tid"] = None
if request_client is None:
return
if stranger_thread:
kind = request_client_kind.get("value", "openai")
if kind == "anthropic_messages":
if stranger_thread:
agent._abort_request_anthropic_client(request_client, reason=reason)
else:
agent._close_request_anthropic_client(request_client, reason=reason)
elif stranger_thread:
agent._abort_request_openai_client(request_client, reason=reason)
else:
agent._close_request_openai_client(request_client, reason=reason)
@ -2976,13 +3011,18 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
usage=usage_obj,
)
def _call_anthropic():
def _call_anthropic(request_client):
"""Stream an Anthropic Messages API response.
Fires delta callbacks for real-time token delivery, but returns
the native Anthropic Message object from get_final_message() so
the rest of the agent loop (validation, tool extraction, etc.)
works unchanged.
Uses ``request_client`` (a per-request Anthropic client registered with
the stranger-thread abort machinery) rather than the shared
``_anthropic_client``, so the stale/interrupt watchdog can abort this
stream's socket without closing the shared client mid-flight (#67142).
"""
has_tool_use = False
# Zero-event guard parity with the chat_completions path: track
@ -3011,7 +3051,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
api_kwargs, log_prefix=getattr(agent, "log_prefix", "")
)
# Use the Anthropic SDK's streaming context manager
with agent._anthropic_client.messages.stream(**api_kwargs) as stream:
with request_client.messages.stream(**api_kwargs) as stream:
# The Anthropic SDK exposes the raw httpx response on
# ``stream.response``. Snapshot diagnostic headers
# immediately so they survive a stream that dies before the
@ -3149,8 +3189,16 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
raise InterruptedError("Agent interrupted before stream retry")
try:
if agent.api_mode == "anthropic_messages":
agent._try_refresh_anthropic_client_credentials()
result["response"] = _call_anthropic()
# #67142: per-request client (credential refresh happens
# inside _create_request_anthropic_client) registered so
# the watchdog aborts its socket, not the shared client.
request_client = _set_request_client(
agent._create_request_anthropic_client(
reason="anthropic_stream_request"
),
kind="anthropic_messages",
)
result["response"] = _call_anthropic(request_client)
else:
result["response"] = _call_chat_completions(stream_attempt_id)
return # success
@ -3276,13 +3324,11 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
)
_cancel_current_stream_attempt("stream_mid_tool_retry_cleanup")
_close_request_client_once("stream_mid_tool_retry_cleanup")
if agent.api_mode == "anthropic_messages":
try:
agent._anthropic_client.close()
agent._rebuild_anthropic_client()
except Exception:
pass
else:
# #67142: anthropic streams on a request-local client,
# already worker-owned-closed by _close_request_client_once
# above; the next attempt builds a fresh one. The shared
# _anthropic_client is never closed from inside a request.
if agent.api_mode != "anthropic_messages":
try:
agent._replace_primary_openai_client(
reason="stream_mid_tool_retry_pool_cleanup"
@ -3341,15 +3387,13 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
# Close the stale request client before retry
_cancel_current_stream_attempt("stream_retry_cleanup")
_close_request_client_once("stream_retry_cleanup")
# Also rebuild the primary client to purge
# any dead connections from the pool.
if agent.api_mode == "anthropic_messages":
try:
agent._anthropic_client.close()
agent._rebuild_anthropic_client()
except Exception:
pass
else:
# Also rebuild the primary client to purge any dead
# connections from the pool. #67142: anthropic uses a
# request-local client (already worker-owned-closed
# above; next attempt builds fresh), so the shared
# _anthropic_client is never closed from inside a
# request — only the OpenAI-wire primary is refreshed.
if agent.api_mode != "anthropic_messages":
try:
agent._replace_primary_openai_client(
reason="stream_retry_pool_cleanup"
@ -3589,11 +3633,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
# Rebuild the primary client too — its connection pool
# may hold dead sockets from the same provider outage.
if agent.api_mode == "anthropic_messages":
try:
agent._anthropic_client.close()
agent._rebuild_anthropic_client()
except Exception:
pass
# #67142: the stale stream ran on a request-local anthropic
# client, already socket-aborted above via
# _close_request_client_once (which unblocks the worker and
# preserves the #28161 no-hang guarantee). The shared
# _anthropic_client is NOT the in-flight transport, so we must
# not close it from this poll (stranger) thread — that was the
# FD-recycle corruption vector. Nothing further is needed.
pass
else:
try:
agent._replace_primary_openai_client(reason="stale_stream_pool_cleanup")
@ -3622,11 +3669,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
)
try:
_cancel_current_stream_attempt("stream_interrupt_abort")
if agent.api_mode == "anthropic_messages":
agent._anthropic_client.close()
agent._rebuild_anthropic_client()
else:
_close_request_client_once("stream_interrupt_abort")
# #67142: kind-aware — anthropic aborts the request-local
# client's socket from this poll thread; the shared
# _anthropic_client is never closed here.
_close_request_client_once("stream_interrupt_abort")
except Exception:
pass
raise InterruptedError("Agent interrupted during streaming API call")

106
run_agent.py
View file

@ -4259,6 +4259,103 @@ class AIAgent:
exc,
)
def _create_request_anthropic_client(self, *, reason: str) -> Any:
"""Build a request-local Anthropic client for one in-flight call.
The shared ``_anthropic_client`` stays the long-lived primary, but the
stale/interrupt watchdog runs on the poll thread and must never call
``close()`` on the client whose TLS socket a worker thread is still
reading: releasing that FD from a stranger thread lets the kernel
recycle it under a still-live SSL BIO, which then writes a TLS record
into an unrelated SQLite header (#29507 / #67142). A per-request client
lets the stranger thread ``shutdown()`` the socket while the owning
worker performs the SDK-level close from its own context — the same
ownership contract the OpenAI-wire path already uses.
Mirrors ``_rebuild_anthropic_client`` construction (direct + Bedrock,
1M-beta drop) but returns a fresh client instead of swapping the shared
one.
"""
if self.api_mode == "anthropic_messages":
self._try_refresh_anthropic_client_credentials()
_drop_1m = bool(getattr(self, "_oauth_1m_beta_disabled", False))
if getattr(self, "provider", None) == "bedrock":
from agent.anthropic_adapter import build_anthropic_bedrock_client
region = getattr(self, "_bedrock_region", "us-east-1") or "us-east-1"
client = build_anthropic_bedrock_client(region)
else:
from agent.anthropic_adapter import build_anthropic_client
client = build_anthropic_client(
self._anthropic_api_key,
getattr(self, "_anthropic_base_url", None),
timeout=get_provider_request_timeout(self.provider, self.model),
drop_context_1m_beta=_drop_1m,
)
logger.debug(
"Anthropic request client created (%s, shared=False) provider=%s model=%s",
reason,
getattr(self, "provider", None),
getattr(self, "model", None),
)
return client
def _close_request_anthropic_client(self, client: Any, *, reason: str) -> None:
"""Owner-thread full close of a request-local Anthropic client.
Force-closes the pool's TCP sockets first (CLOSE-WAIT hygiene, parity
with ``_close_openai_client``), then does the graceful SDK close. Safe
because the caller owns the connection.
"""
if client is None:
return
try:
self._force_close_tcp_sockets(client)
client.close()
logger.info(
"Anthropic client closed (%s, shared=False) provider=%s model=%s",
reason,
getattr(self, "provider", None),
getattr(self, "model", None),
)
except Exception as exc:
logger.debug(
"Anthropic client close failed (%s, shared=False) provider=%s model=%s error=%s",
reason,
getattr(self, "provider", None),
getattr(self, "model", None),
exc,
)
def _abort_request_anthropic_client(self, client: Any, *, reason: str) -> None:
"""Cross-thread abort for request-local Anthropic clients.
Stranger threads (the interrupt-check / stale-stream detector loop)
must not call the SDK ``close()`` — that races the owning worker's live
SSL BIO and can recycle a TLS FD into a SQLite header (#29507 /
#67142). Only ``shutdown(SHUT_RDWR)`` the pool's sockets so the worker
unblocks and releases the FD from its own thread.
"""
if client is None:
return
try:
shutdown_count = self._force_close_tcp_sockets(client)
logger.info(
"Anthropic client aborted (%s, shared=False, tcp_force_closed=%d, "
"deferred_close=stranger_thread) provider=%s model=%s",
reason,
shutdown_count,
getattr(self, "provider", None),
getattr(self, "model", None),
)
except Exception as exc:
logger.debug(
"Anthropic client abort failed (%s, shared=False) provider=%s model=%s error=%s",
reason,
getattr(self, "provider", None),
getattr(self, "model", None),
exc,
)
def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta: callable = None):
"""Forwarder — see ``agent.codex_runtime.run_codex_stream``."""
from agent.codex_runtime import run_codex_stream
@ -4644,15 +4741,18 @@ class AIAgent:
return False
return pool.has_available()
def _anthropic_messages_create(self, api_kwargs: dict):
if self.api_mode == "anthropic_messages":
def _anthropic_messages_create(self, api_kwargs: dict, *, client: Any = None):
# When a request-local client is supplied it was already credential-
# refreshed in ``_create_request_anthropic_client``; only the shared
# fallback path refreshes here.
if client is None and self.api_mode == "anthropic_messages":
self._try_refresh_anthropic_client_credentials()
# Defensive: strip Responses-only kwargs that can leak in under an
# api_mode-flip race (the Anthropic SDK raises a non-retryable
# TypeError on them). See #31673.
from agent.anthropic_adapter import create_anthropic_message
return create_anthropic_message(
self._anthropic_client,
client or self._anthropic_client,
api_kwargs,
log_prefix=getattr(self, "log_prefix", ""),
prefer_stream=not bool(getattr(self, "_disable_streaming", False)),

90
tests/agent/test_cascading_interrupt_6600.py
View file

@ -132,3 +132,93 @@ def test_request_cancelled_token_is_request_local():
with pytest.raises(httpx.RemoteProtocolError):
cch.interruptible_api_call(agent, {"model": "x", "messages": []})
# ---------------------------------------------------------------------------
# #67142: direct-Anthropic stale/interrupt watchdog must abort the request-local
# client from the poll (stranger) thread and NEVER close/rebuild the shared
# _anthropic_client — closing it there released a live TLS FD that the kernel
# recycled into a SQLite handle, writing a TLS record over a DB header.
# ---------------------------------------------------------------------------
def _make_anthropic_agent():
agent = _make_agent()
agent.api_mode = "anthropic_messages"
return agent
def _wait_for_mock_call(mock, timeout=3.0):
deadline = time.time() + timeout
while time.time() < deadline:
if mock.called:
return
time.sleep(0.02)
raise AssertionError(f"{mock!r} was not called within {timeout}s")
def test_anthropic_non_streaming_stale_aborts_request_client_not_shared():
"""Stale non-streaming Anthropic call: the poll thread aborts the
request-local client's socket; the shared client is never closed/rebuilt,
and the worker still unblocks and closes its own client (no #28161 hang)."""
agent = _make_anthropic_agent()
agent._compute_non_stream_stale_timeout.return_value = 0.05
agent._codex_silent_hang_hint = MagicMock(return_value=None)
request_client = MagicMock()
agent._create_request_anthropic_client = MagicMock(return_value=request_client)
agent._abort_request_anthropic_client = MagicMock()
agent._close_request_anthropic_client = MagicMock()
def _create(_api_kwargs, *, client):
assert client is request_client
# Outlive the 0.05s stale timeout AND the worker join (2.0s) so the
# stale detector surfaces its TimeoutError.
time.sleep(2.5)
return object()
agent._anthropic_messages_create = MagicMock(side_effect=_create)
with pytest.raises(TimeoutError):
cch.interruptible_api_call(agent, {"model": "x", "messages": []})
# Shared client untouched from the poll thread.
agent._anthropic_client.close.assert_not_called()
agent._rebuild_anthropic_client.assert_not_called()
# Poll (stranger) thread aborts the request-local client's socket only.
agent._abort_request_anthropic_client.assert_called_once_with(
request_client, reason="stale_call_kill"
)
# Worker unblocks and closes its own request client from its own thread.
_wait_for_mock_call(agent._close_request_anthropic_client)
def test_anthropic_non_streaming_interrupt_aborts_request_client_not_shared():
"""Interrupted non-streaming Anthropic call: near-instant InterruptedError,
request-local client aborted from the poll thread, shared client untouched."""
agent = _make_anthropic_agent()
request_client = MagicMock()
agent._create_request_anthropic_client = MagicMock(return_value=request_client)
agent._abort_request_anthropic_client = MagicMock()
agent._close_request_anthropic_client = MagicMock()
def _create(_api_kwargs, *, client):
assert client is request_client
agent._interrupt_requested = True
time.sleep(1.0)
raise httpx.RemoteProtocolError("forced close would have happened")
agent._anthropic_messages_create = MagicMock(side_effect=_create)
t0 = time.time()
with pytest.raises(InterruptedError):
cch.interruptible_api_call(agent, {"model": "x", "messages": []})
elapsed = time.time() - t0
assert elapsed < 3.0, f"interrupt took {elapsed:.1f}s — should be near-instant"
agent._anthropic_client.close.assert_not_called()
agent._rebuild_anthropic_client.assert_not_called()
agent._abort_request_anthropic_client.assert_called_once_with(
request_client, reason="interrupt_abort"
)

70
tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py
View file

@ -1,16 +1,24 @@
"""Anthropic stream cleanup must call _anthropic_client.close() + _rebuild_anthropic_client(),
not _replace_primary_openai_client(), to avoid 15-minute hangs on Anthropic-native configs.
"""Anthropic stream cleanup must not call _replace_primary_openai_client() and
must not hang on Anthropic-native configs (#28161), now via the request-local
client model (#67142).
Three cleanup sites in chat_completion_helpers.interruptible_streaming_api_call() were
calling _replace_primary_openai_client() unconditionally. For api_mode=anthropic_messages
this silently fails (no OPENAI_API_KEY) and leaves the in-flight httpx stream unclosed,
blocking the worker thread until the 900s httpx read-timeout fires.
Originally three cleanup sites in interruptible_streaming_api_call() called
_replace_primary_openai_client() unconditionally; for api_mode=anthropic_messages
that silently failed (no OPENAI_API_KEY) and left the in-flight httpx stream
unclosed, blocking the worker until the 900s read-timeout fired.
Since #67142, anthropic streams run on a per-request client: the stale/retry
cleanup closes the *request-local* client (worker-owned) and builds a fresh one
next attempt — the shared _anthropic_client is never closed/rebuilt from inside
a request (that poll-thread close was the TLS-FD→SQLite corruption vector). The
no-hang guarantee is preserved because the poll thread aborts the request
client's sockets, which unblocks the worker.
Tests cover:
- stream_retry_pool_cleanup (connection error on fresh stream, L1836)
- stale_stream_pool_cleanup (outer poll loop detects stale stream, L1987)
- stream_retry cleanup (connection error on fresh stream)
- stale_stream cleanup (outer poll loop detects stale stream)
Fixes #28161
Fixes #28161. Extends #67142.
"""
import threading
from types import SimpleNamespace
@ -41,6 +49,9 @@ def _make_anthropic_agent(**kwargs):
agent.api_mode = "anthropic_messages"
agent._anthropic_client = MagicMock()
agent._anthropic_api_key = "test-anthropic-key"
# #67142: anthropic streams now run on a request-local client; route it to
# the test mock so .messages.stream is exercised and its cleanup observed.
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
return agent
@ -74,13 +85,16 @@ def _failing_stream_cm():
class TestAnthropicStreamPoolCleanup:
"""_replace_primary_openai_client must not be called for api_mode=anthropic_messages."""
"""Anthropic cleanup must never touch the OpenAI primary or the shared
Anthropic client, and must not hang (#28161 / #67142)."""
@pytest.mark.filterwarnings(
"ignore::pytest.PytestUnhandledThreadExceptionWarning"
)
def test_stream_retry_calls_anthropic_rebuild_not_openai(self):
"""Connection error during stream retry → close+rebuild Anthropic client, not OpenAI."""
def test_stream_retry_closes_request_client_not_openai(self):
"""Connection error during stream retry → close the request-local
Anthropic client (worker-owned) and retry; never rebuild the shared
Anthropic client, never touch the OpenAI primary."""
agent = _make_anthropic_agent()
attempt_count = [0]
@ -100,14 +114,19 @@ class TestAnthropicStreamPoolCleanup:
agent._interruptible_streaming_api_call({})
mock_replace.assert_not_called()
mock_rebuild.assert_called_once()
agent._anthropic_client.close.assert_called_once()
# #67142: the shared client is never rebuilt from inside a request; the
# request-local client (routed to this mock) is closed instead.
mock_rebuild.assert_not_called()
agent._anthropic_client.close.assert_called()
assert attempt_count[0] == 2 # retried once, then succeeded
@pytest.mark.filterwarnings(
"ignore::pytest.PytestUnhandledThreadExceptionWarning"
)
def test_stale_stream_calls_anthropic_rebuild_not_openai(self, monkeypatch):
"""Stale-stream outer-poll detector → close+rebuild Anthropic client, not OpenAI."""
def test_stale_stream_aborts_request_client_not_openai(self, monkeypatch):
"""Stale-stream outer-poll detector → abort the request-local client's
socket (unblocking the worker) and retry; never _replace_primary_openai
and never rebuild the shared Anthropic client."""
monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "0.1")
agent = _make_anthropic_agent()
@ -117,14 +136,15 @@ class TestAnthropicStreamPoolCleanup:
def _stream_side_effect(*args, **kwargs):
attempt_count[0] += 1
if attempt_count[0] == 1:
# First attempt: stream that yields nothing (triggers stale detector),
# then raises ConnectError once _anthropic_client.close() unblocks it.
# First attempt: stream that yields nothing (triggers stale
# detector), then raises ConnectError once the poll thread
# aborts the request client's socket.
cm = MagicMock()
stream = MagicMock()
def _blocking_gen():
unblock.wait(timeout=5.0)
raise httpx.ConnectError("connection dropped after close()")
raise httpx.ConnectError("connection dropped after abort")
yield # make this a generator so next() triggers the wait
stream.__iter__ = MagicMock(return_value=_blocking_gen())
@ -135,8 +155,10 @@ class TestAnthropicStreamPoolCleanup:
return _good_stream_cm()
agent._anthropic_client.messages.stream.side_effect = _stream_side_effect
# close() on the mock Anthropic client unblocks the inner thread.
agent._anthropic_client.close.side_effect = unblock.set
# #67142: the stale detector aborts the request-local client's sockets
# from the poll thread (not close() on the shared client); simulate the
# socket shutdown waking the blocked read.
agent._abort_request_anthropic_client = lambda *a, **k: unblock.set()
with patch.object(agent, "_rebuild_anthropic_client") as mock_rebuild:
with patch.object(
@ -145,6 +167,6 @@ class TestAnthropicStreamPoolCleanup:
agent._interruptible_streaming_api_call({})
mock_replace.assert_not_called()
# close() and rebuild called at least once by the stale detector.
agent._anthropic_client.close.assert_called()
assert mock_rebuild.call_count >= 1
# The shared Anthropic client is never rebuilt from inside a request.
mock_rebuild.assert_not_called()
assert attempt_count[0] >= 2 # stale-killed once, then retried

59
tests/run_agent/test_run_agent.py
View file

@ -7648,13 +7648,60 @@ class TestAnthropicInterruptHandler:
assert "anthropic_messages" in source, \
"interruptible_api_call must handle Anthropic interrupt (api_mode check)"
def test_interruptible_rebuilds_anthropic_client(self):
"""After interrupting, the Anthropic client should be rebuilt."""
import inspect
def test_interruptible_anthropic_interrupt_never_closes_shared_client(self):
"""#67142: a non-streaming Anthropic interrupt must abort the
request-local client from the poll thread, never close/rebuild the
shared _anthropic_client (which raced a live SSL BIO and corrupted an
unrelated SQLite DB via TLS-FD recycling).
Replaces the former source-reading assertion (which asserted the old,
now-removed rebuild-on-interrupt behavior) with a behavior test.
"""
import threading
import time
from unittest.mock import MagicMock
from run_agent import AIAgent
from agent.chat_completion_helpers import interruptible_api_call
source = inspect.getsource(interruptible_api_call)
assert "build_anthropic_client" in source, \
"interruptible_api_call must rebuild Anthropic client after interrupt"
agent = AIAgent(
api_key="test-key",
base_url="https://api.anthropic.com",
provider="anthropic",
model="claude-test",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
agent.api_mode = "anthropic_messages"
agent._interrupt_requested = False
agent._anthropic_client = MagicMock()
agent._rebuild_anthropic_client = MagicMock()
request_client = MagicMock()
agent._create_request_anthropic_client = MagicMock(return_value=request_client)
agent._abort_request_anthropic_client = MagicMock()
agent._close_request_anthropic_client = MagicMock()
def _create(_api_kwargs, *, client):
assert client is request_client
agent._interrupt_requested = True
time.sleep(1.0)
raise RuntimeError("forced close would have happened")
agent._anthropic_messages_create = MagicMock(side_effect=_create)
t0 = time.time()
with pytest.raises(InterruptedError):
interruptible_api_call(agent, {"model": "x", "messages": []})
elapsed = time.time() - t0
assert elapsed < 3.0, f"interrupt took {elapsed:.1f}s — should be near-instant"
# The shared client is never closed/rebuilt from the poll thread.
agent._anthropic_client.close.assert_not_called()
agent._rebuild_anthropic_client.assert_not_called()
# The poll (stranger) thread aborts the request-local client's socket.
agent._abort_request_anthropic_client.assert_called_once_with(
request_client, reason="interrupt_abort"
)
def test_streaming_has_anthropic_branch(self):
"""_streaming_api_call must also handle Anthropic interrupt."""

8
tests/run_agent/test_stream_stale_circuit_breaker.py
View file

@ -39,6 +39,9 @@ def _make_anthropic_agent(**kwargs):
agent.api_mode = "anthropic_messages"
agent._anthropic_client = MagicMock()
agent._anthropic_api_key = "test-anthropic-key"
# #67142: anthropic streams now run on a request-local client; route it to
# the test mock so .messages.stream is exercised.
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
return agent
@ -116,7 +119,10 @@ class TestStreamStaleCircuitBreaker:
# Every attempt blocks, trips the stale detector, and fails.
agent._anthropic_client.messages.stream.side_effect = _stream_side_effect
agent._anthropic_client.close.side_effect = unblock.set
# #67142: the stale detector now aborts the request-local client's
# sockets from the poll thread (not close() on the shared client), so
# unblock on the abort to simulate the socket shutdown waking the read.
agent._abort_request_anthropic_client = lambda *a, **k: unblock.set()
with pytest.raises(Exception):
agent._interruptible_streaming_api_call({})

25
tests/run_agent/test_streaming.py
View file

@ -1196,6 +1196,9 @@ class TestAnthropicStreamCallbacks:
agent._anthropic_client = MagicMock()
agent._anthropic_client.messages.stream.return_value = mock_stream
# #67142: streaming now runs on a request-local anthropic client; route
# it to the test mock so .messages.stream is exercised.
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
agent._interruptible_streaming_api_call({})
@ -1251,16 +1254,18 @@ class TestAnthropicStreamCallbacks:
_BadStream(),
good_stream,
]
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
response = agent._interruptible_streaming_api_call({})
assert response is final_message
assert agent._anthropic_client.messages.stream.call_count == 2
# Anthropic-native cleanup: close + rebuild the Anthropic client, never
# the OpenAI primary client.
# #67142: cleanup runs on the request-local anthropic client (closed,
# worker-owned, via _close_request_client_once), never rebuilding the
# shared client and never touching the OpenAI primary client.
assert mock_replace.call_count == 0
assert mock_rebuild.call_count == 1
assert agent._anthropic_client.close.call_count == 1
assert mock_rebuild.call_count == 0
assert agent._anthropic_client.close.call_count >= 1
@patch("run_agent.AIAgent._replace_primary_openai_client")
def test_generic_anthropic_valueerror_still_propagates_without_stream_retry(
@ -1286,6 +1291,7 @@ class TestAnthropicStreamCallbacks:
agent._anthropic_client.messages.stream.side_effect = ValueError(
"invalid local request shape"
)
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
with pytest.raises(ValueError, match="invalid local request shape"):
agent._interruptible_streaming_api_call({})
@ -1326,15 +1332,17 @@ class TestAnthropicStreamCallbacks:
agent._anthropic_client = MagicMock()
agent._anthropic_client.messages.stream.return_value = empty_stream
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
with pytest.raises(EmptyStreamError):
agent._interruptible_streaming_api_call({})
assert agent._anthropic_client.messages.stream.call_count == 3
# Anthropic-native cleanup between attempts: rebuild the Anthropic
# client, never the OpenAI primary client.
# #67142: cleanup between attempts runs on the request-local anthropic
# client (fresh one built per attempt); the shared client is never
# rebuilt and the OpenAI primary client is never touched.
assert mock_replace.call_count == 0
assert mock_rebuild.call_count == 2
assert mock_rebuild.call_count == 0
@patch("run_agent.AIAgent._try_refresh_anthropic_client_credentials")
@patch("run_agent.AIAgent._rebuild_anthropic_client")
@ -1369,13 +1377,14 @@ class TestAnthropicStreamCallbacks:
agent._anthropic_client = MagicMock()
agent._anthropic_client.messages.stream.return_value = empty_stream
agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
with pytest.raises(EmptyStreamError):
agent._interruptible_streaming_api_call({})
assert agent._anthropic_client.messages.stream.call_count == 3
assert mock_replace.call_count == 0
assert mock_rebuild.call_count == 2
assert mock_rebuild.call_count == 0
class TestPartialToolCallWarning: