# NousResearch/hermes-agent · TLS FD-recycling corruption of SQLite databases ## Open now - Nothing open on GitHub. These issues and fixes concern a recurring class of bug where interrupted or stale HTTP/TLS connections have their file descriptors recycled and subsequently written into open SQLite database files. Multiple vectors and attempted fixes are represented across Anthropic and OpenAI client paths. ## TL;DR - 9 member(s): 0 open issue(s), 3 closed issue(s), 0 open PR(s), 2 merged PR(s), 4 PR(s) closed without merging. - A fix was merged to main in PR #30858. Verify it addresses your case. - **Recommended:** read PR #30858 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-30858,pr-67238`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text. - Members outside that structural core may belong to a different underlying problem. - Pairwise link density `0.22`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #29507: Interrupted OpenAI/httpx request thread survives across turns and writes TLS record bytes to unrelated file descriptors on delayed close - #44037: Recurring fd-recycle corruption: TLS application-data records written into memory_store.db while gateway holds it open (same class as #29507, new vector) - #67142: Anthropic stale-stream watchdog can still corrupt SQLite via TLS FD reuse ## Prior & competing PRs - **#30858: fix(api-call): close FD-recycling race that wrote TLS bytes into kanban.db (#29507)**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `agent/agent_runtime_helpers.py`, `agent/chat_completion_helpers.py`, `run_agent.py`, `tests/run_agent/test_create_openai_client_reuse.py`, `tests/run_agent/test_tls_fd_recycle_corruption.py` - fix hunks: - `agent/agent_runtime_helpers.py` ```diff @@ -2132,33 +2132,56 @@ def apply_pending_steer_to_tool_results(agent, messages: list, num_tool_msgs: in def force_close_tcp_sockets(client: Any) -> int: - """Force-close underlying TCP sockets to prevent CLOSE-WAIT accumulation. - - When a provider drops a connection mid-stream, httpx's ``client.close()`` - performs a graceful shutdown which leaves sockets in CLOSE-WAIT until the - OS times them out (often minutes). This method walks the httpx transport - pool and issues ``socket.shutdown(SHUT_RDWR)`` + ``socket.close()`` to - force an immediate TCP RST, freeing the file descriptors. - - Returns the number of sockets force-closed. + """Abort in-flight TCP I/O by shutting down sockets WITHOUT closing FDs. + + When a provider drops a connection mid-stream — or the user issues an + interrupt — we want to unblock httpx's reader/writer immediately rather + than waiting for the kernel's per-connection timeout. ``shutdown(SHUT_RDWR)`` + achieves that: it sends FIN, breaks any pending ``recv``/``send`` with EOF + or ``EPIPE``, but does NOT release the file descriptor. + + Historically this helper also called ``socket.close()`` so the FD got + released immediately, but that's unsafe when (as is the case for both the + interrupt-abort path and stale-call kill path) the helper runs on a + different thread than the one driving the request: + + * The Python ``socket.socket`` we close here is the SAME object held by + httpx's pool, so closing it via Python sets its ``_fd`` to -1 and + future operations on that Python object fail safely. + * BUT the SSL wrapper (``ssl.SSLSocket``'s underlying OpenSSL ``BIO``) + caches the raw integer FD. Once ``os.close(fd)`` runs, the kernel may + immediately recycle that integer to the next ``open()`` call — e.g. + the kanban dispatcher opening ``kanban.db``. + * The owning worker thread then unwinds httpx, the SSL layer flushes a + pending TLS record, and the encrypted bytes get written into the + wrong file (issue #29507: 24-byte TLS application-data record + clobbering SQLite header bytes 5..28). + + The fix is to let the owning thread own the close. ``shutdown()`` from any + thread is FD-safe; ``close()`` is not. The httpx connection's own close + path — which runs from the worker thread when it unwinds — will release + the FD via the same ``socket.socket`` object, and because Python's socket + close atomically swaps ``_fd`` to -1 *before* issuing ``os.close``, there + is no FD-aliasing window when only one thread closes. + + Returns the number of sockets shut down. (Field kept as + ``tcp_force_closed=N`` in the log line for backwards-compatible parsing.) """ import socket as _socket - closed = 0 + shutdown_count = 0 try: for sock in _iter_pool_sockets(client): try: sock.shutdown(_socket.SHUT_RDWR) except OSError: + # Already shut down / not connected / FD invalid — all benign. pass - try: - sock.close() - except OSError: - pass - closed += 1 + # IMPORTANT (#29507): do NOT call sock.close() here. See docstring. + shutdown_count += 1 except Exception as exc: _ra().logger.debug("Force-close TCP sockets sweep error: %s", exc) - return closed + return shutdown_count ``` - `agent/chat_completion_helpers.py` ```diff @@ -91,23 +91,55 @@ def interruptible_api_call(agent, api_kwargs: dict): provider fallback. """ result = {"response": None, "error": None} - request_client_holder = {"client": None} + request_client_holder = {"client": None, "owner_tid": None} request_client_lock = threading.Lock() def _set_request_client(client): with request_client_lock: request_client_holder["client"] = client + # #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()``. + request_client_holder["owner_tid"] = threading.get_ident() return client def _take_request_client(): with request_client_lock: client = request_client_holder.get("client") request_client_holder["client"] = None + request_client_holder["owner_tid"] = None return client def _close_request_client_once(reason: str) -> None: - request_client = _take_request_client() - if request_client is not None: + # #29507: dispatch on the calling thread. + # + # When ``_call`` (the worker) reaches its ``finally`` it owns the + # close and we pop + fully close as before. When a *stranger* thread + # (the interrupt-check loop, the stale-call detector) drives the + # close, only shut the sockets down so the worker's blocked + # ``recv``/``send`` unwinds with an ``EPIPE`` / EOF — and let the + # worker close ``client`` from its own thread on its way out. That + # avoids the FD-recycling race where the kernel reassigned a + # just-closed TLS socket FD to ``kanban.db``, and the still-live SSL + # BIO on the worker thread then wrote a 24-byte TLS application-data + # record into the SQLite header (#29507). + with request_client_lock: + request_client = request_client_holder.get("client") + owner_tid = request_client_holder.get("owner_tid") + stranger_thread = ( + request_client is not None + and owner_tid is not None + and owner_tid != threading.get_ident() + ) + if not stranger_thread: + # Owning thread (or no recorded owner) → pop and fully close. + request_client_holder["client"] = None + request_client_holder["owner_tid"] = None + if request_client is None: + return + if stranger_thread: + agent._abort_request_openai_client(request_client, reason=reason) + else: agent._close_request_openai_client(request_client, reason=reason) def _call(): ``` ```diff @@ -1271,23 +1303,44 @@ def _on_reasoning(text): return result["response"] result = {"response": None, "error": None, "partial_tool_names": []} - request_client_holder = {"client": None, "diag": None} + request_client_holder = {"client": None, "diag": None, "owner_tid": None} request_client_lock = threading.Lock() def _set_request_client(client): with request_client_lock: request_client_holder["client"] = client + # See #29507 explanation in the non-streaming variant above. + request_client_holder["owner_tid"] = threading.get_ident() return client def _take_request_client(): with request_client_lock: client = request_client_holder.get("client") request_client_holder["client"] = None + request_client_holder["owner_tid"] = None return client def _close_request_client_once(reason: str) -> None: - request_client = _take_request_client() - if request_client is not None: + # See #29507 explanation in the non-streaming variant above. A + # stranger thread (the interrupt-check / stale-stream detector loop) + # only aborts sockets — never pops, never calls ``client.close()`` — + # so the worker thread retains ownership of the FD release. + with request_client_lock: + request_client = request_client_holder.get("client") + owner_tid = request_client_holder.get("owner_tid") + stranger_thread = ( + request_client is not None + and owner_tid is not None + and owner_tid != threading.get_ident() + ) + if not stranger_thread: + request_client_holder["client"] = None + request_client_holder["owner_tid"] = None + if request_client is None: + return + if stranger_thread: + agent._abort_request_openai_client(request_client, reason=reason) + else: agent._close_request_openai_client(request_client, reason=reason) first_delta_fired = {"done": False} ``` - `run_agent.py` ```diff @@ -2563,6 +2563,39 @@ def _create_request_openai_client(self, *, reason: str, api_kwargs: Optional[dic def _close_request_openai_client(self, client: Any, *, reason: str) -> None: self._close_openai_client(client, reason=reason, shared=False) + def _abort_request_openai_client(self, client: Any, *, reason: str) -> None: + """Cross-thread abort: shut sockets down without releasing FDs. + + Companion to :meth:`_close_request_openai_client` for stranger-thread + callers (interrupt-check loop, stale-call detector). Calling + ``client.close()`` from a thread that does not own the active httpx + connection raced the still-live SSL BIO and corrupted unrelated file + descriptors when the kernel recycled the just-freed TCP FD (#29507). + + Here we only ``shutdown(SHUT_RDWR)`` the pool's sockets. That unblocks + the owning worker thread's pending ``recv``/``send`` with an EOF or + ``EPIPE`` so it can unwind and close ``client`` from its own context + — which is where the FD release belongs. + """ + if client is None: + return + try: + shutdown_count = self._force_close_tcp_sockets(client) + logger.info( + "OpenAI client aborted (%s, shared=False, tcp_force_closed=%d, " + "deferred_close=stranger_thread) %s", + reason, + shutdown_count, + self._client_log_context(), + ) + except Exception as exc: + logger.debug( + "OpenAI client abort failed (%s, shared=False) %s error=%s", + reason, + self._client_log_context(), + 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 ``` - `tests/run_agent/test_create_openai_client_reuse.py` ```diff @@ -190,7 +190,13 @@ def test_replace_primary_openai_client_survives_repeated_rebuilds(): def test_force_close_tcp_sockets_descends_httpcore_1_connection_wrapper(): - """httpcore 1.x stores the real stream below conn._connection.""" + """httpcore 1.x stores the real stream below conn._connection. + + Post-#29507: the helper must shut sockets down but must NOT release the + FD via ``sock.close()`` — that race recycled FDs into unrelated file + descriptors (kanban.db) and let TLS bytes overwrite SQLite headers. The + owning httpx thread is responsible for closing FDs on its own unwind. + """ from agent.agent_runtime_helpers import force_close_tcp_sockets class FakeSocket: ``` ```diff @@ -215,4 +221,6 @@ def close(self): assert force_close_tcp_sockets(openai_client) == 1 assert sock.shutdown_calls == 1 - assert sock.close_calls == 1 + # #29507: close() must NOT be called from this helper — the owning + # httpx worker thread releases the FD, not us. + assert sock.close_calls == 0 ``` - `tests/run_agent/test_tls_fd_recycle_corruption.py` ```diff @@ -0,0 +1,454 @@ +"""Regressions for issue #29507 — cross-thread close of the per-request OpenAI +client could release a TLS socket FD whose integer was still cached in the +owning httpx worker's SSL BIO. The kernel then recycled the FD into the next +``open()`` (e.g. the kanban dispatcher's ``kanban.db``), and the worker's +delayed TLS flush wrote a 24-byte TLS application-data record on top of the +SQLite header. + +The fix has two prongs: + +1. ``force_close_tcp_sockets`` no longer calls ``sock.close()`` — only + ``shutdown(SHUT_RDWR)``. Shutdown unblocks the worker's pending + ``recv``/``send`` without releasing the FD. + +2. ``_close_request_client_once`` is thread-aware: a stranger thread (the + interrupt-check / stale-call loop) only aborts the sockets and leaves + the client in the holder; the worker's own ``finally`` performs the + actual ``client.close()`` from its own thread context. + +Both prongs together close the FD-recycling window. The tests below pin +each prong individually and one end-to-end test simulates the reporter's +timeline at object granularity (no network, no real sockets). +""" +from __future__ import annotations + +import logging +import socket as _socket +import threading +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Prong 1: force_close_tcp_sockets must NOT release file descriptors. +# --------------------------------------------------------------------------- + + +class _FakeSocket: + """Records shutdown/close calls without touching real FDs.""" + + def __init__(self): + self.shutdown_calls = 0 + self.close_calls = 0 + + def shutdown(self, _how): + self.shutdown_calls += 1 + + def close(self): + self.close_calls += 1 + + +def _build_fake_client(sock): + """Mimic the httpcore-1 layout that ``_iter_pool_sockets`` walks.""" + stream = SimpleNamespace(_sock=sock) + http11 = SimpleNamespace(_network_stream=stream) + pool_entry = SimpleNamespace(_connection=http11) + pool = SimpleNamespace(_connections=[pool_entry]) + transport = SimpleNamespace(_pool=pool) + http_client = SimpleNamespace(_transport=transport) + return SimpleNamespace(_client=http_client) + + +def test_force_close_tcp_sockets_shutdown_only_no_close(): + """The smoking-gun guarantee: shutdown is called, close is NOT. + + If a future refactor reintroduces ``sock.close()`` here, the + FD-recycling race that corrupted ``kanban.db`` (issue #29507) will + re-open. Pin the contract explicitly. + """ + from agent.agent_runtime_helpers import force_close_tcp_sockets + + sock = _FakeSocket() + client = _build_fake_client(sock) + + n = force_close_tcp_sockets(client) + + assert n == 1 + assert sock.shutdown_calls == 1, "shutdown() must run — it's how we unblock the worker" + assert sock.close_calls == 0, ( + "close() must NOT run from this helper — releasing the FD here is the " + "race that wrote TLS bytes into kanban.db (#29507)" + ) + + +def test_force_close_tcp_sockets_uses_shut_rdwr(): + """Both directions must be shut down so the SSL state machine fully unwinds. + + Half-close (e.g. SHUT_WR only) wouldn't unblock a worker blocked in + ``recv``, defeating the whole point of the helper. + """ + from agent.agent_runtime_helpers import force_close_tcp_sockets + + captured = [] + + class _ProbingSocket: + def shutdown(self, how): + captured.append(how) + + def close(self): # pragma: no cover — must not run, asserted below + captured.append("CLOSE_CALLED") + + sock = _ProbingSocket() + client = _build_fake_client(sock) + + force_close_tcp_sockets(client) + + assert captured == [_socket.SHUT_RDWR] + + +def test_force_close_tcp_sockets_swallows_oserror_on_shutdown(): + """A socket already shut down / not connected raises ``OSError`` — benign.""" + from agent.agent_runtime_helpers import force_close_tcp_sockets + + class _AlreadyShut: + def shutdown(self, _how): + raise OSError("not connected") + + def close(self): # pragma: no cover — must not run + raise AssertionError("close() must not be called") + + client = _build_fake_client(_AlreadyShut()) + + # No exception escapes; the helper still counts the socket as handled. + assert force_close_tcp_sockets(client) == 1 + + +def test_force_close_tcp_sockets_handles_multiple_pool_entries(): + """Walk every pool connection — the bug equally applies to all of them.""" + from agent.agent_runtime_helpers import force_close_tcp_sockets + + socks = [_FakeSocket(), _FakeSocket(), _FakeSocket()] + entries = [ + SimpleNamespace(_connection=SimpleNamespace(_network_stream=SimpleNamespace(_sock=s))) + for s in socks + ] + pool = SimpleNamespace(_connections=entries) + transport = SimpleNamespace(_pool=pool) + http_client = SimpleNamespace(_transport=transport) + client = SimpleNamespace(_client=http_client) + + assert force_close_tcp_sockets(client) == 3 + for s in socks: + assert s.shutdown_calls == 1 + assert s.close_calls == 0 + + +# --------------------------------------------------------------------------- +# Prong 2: _close_request_client_once is thread-aware. +# --------------------------------------------------------------------------- + + +def _make_agent_mock(): + """Minimal agent with the two close primitives stubbed for spy-style checks.""" + agent = MagicMock() + agent._interrupt_requested = False + agent._close_request_openai_client = MagicMock() + agent._abort_request_openai_client = MagicMock() + return agent + + +def _call_inside_owner_thread(callable_): + """Run callable_ on a separate thread so its ``threading.get_ident()`` + differs from the test thread.""" + result = {"value": None, "exc": None} + + def runner(): + try: + result["value"] = callable_() + except BaseException as e: # noqa: BLE001 — propagate test failures faithfully + result["exc"] = e + + t = threading.Thread(target=runner) + t.start() + t.join(timeout=5.0) + if result["exc"] is not None: + raise result["exc"] + return result["value"] + + +def test_close_from_stranger_thread_aborts_only_no_close(): + """Stranger-thread close → ``_abort_request_openai_client``, holder NOT popped. + + Reproduces the asyncio_0 → Thread-1616 interrupt path. After this call + the worker's eventual ``finally`` must still see the client in the + holder so IT can be the one releasing the FD. + """ + from agent.chat_completion_helpers import interruptible_api_call + + # We can't easily invoke just `_close_request_client_once` because it's + # a closure local to ``interruptible_api_call``. Re-extract the same + # logic by exercising it through a fake worker that lets us drive the + # holder state manually. + agent = _make_agent_mock() + # Pretend ``_call`` ran far enough to set the client on the holder + # from the owner thread. + sentinel = object() + owner_tid_holder = {"tid": None, "client_present_after_stranger_close": False} + + def _owner_workload(holder, lock): + # Owner-thread set + with lock: + holder["client"] = sentinel + holder["owner_tid"] = threading.get_ident() + owner_tid_holder["tid"] = threading.get_ident() + + holder = {"client": None, "owner_tid": None} + lock = threading.Lock() + _call_inside_owner_thread(lambda: _owner_workload(holder, lock)) + + # Now drive the exact body of the post-#29507 ``_close_request_client_once`` + # from the test thread (stranger) and from the owner thread. + def close_once(holder, lock, reason): + with lock: + request_client = holder.get("client") + owner_tid = holder.get("owner_tid") + stranger = ( + request_client is not None + and owner_tid is not None + and owner_tid != threading.get_ident() + ) + if not stranger: + holder["client"] = None + holder["owner_tid"] = None + if request_client is None: + return None + if stranger: + agent._abort_request_openai_client(request_client, reason=reason) + return "aborted" + agent._close_request_openai_client(request_client, reason=reason) + return "closed" + + outcome = close_once(holder, lock, "interrupt_abort") + + assert outcome == "aborted" + agent._abort_request_openai_client.assert_called_once() + agent._close_request_openai_client.assert_not_called() + # Holder is still populated — the worker thread will pick this up in + # its ``finally`` and own the actual ``client.close()``. + assert holder["client"] is sentinel + assert holder["owner_tid"] == owner_tid_holder["tid"] + + +def test_close_from_owner_thread_pops_and_full_close(): + """Worker-thread close → ``_close_request_openai_client``, holder popped.""" + agent = _make_agent_mock() + sentinel = object() + holder = {"client": None, "owner_tid": None} + lock = threading.Lock() + + def workload(): + with lock: + holder["client"] = sentinel + holder["owner_tid"] = threading.get_ident() + + # Same body inlined here so the test thread and the closing thread + # are identical (owner == self). + with lock: + request_client = holder.get("client") + owner_tid = holder.get("owner_tid") + stranger = ( + request_client is not None + and owner_tid is not None + and owner_tid != threading.get_ident() + ) + if not stranger: + holder["client"] = None + holder["owner_tid"] = None + if request_client is None: + return None + if stranger: + agent._abort_request_openai_client(request_client, reason="request_complete") + return "aborted" + agent._close_request_openai_client(request_client, reason="request_complete") + return "closed" + + outcome = _call_inside_owner_thread(workload) + + assert outcome == "closed" + agent._close_request_openai_client.assert_called_once() + agent._abort_request_openai_client.assert_not_called() + assert holder["client"] is None + assert holder["owner_tid"] is None + + +def test_stranger_then_owner_close_sequence_runs_full_close_exactly_once(): + """Stranger abort followed by owner close → full close runs once. + + This mirrors the reporter's timeline: asyncio_0 fires interrupt_abort + (stranger → abort only), then Thread-1616 unwinds and its finally + fires request_complete (owner → full close). Net result must be one + abort + one full close, with the holder ending empty. + """ + agent = _make_agent_mock() + sentinel = object() + holder = {"client": None, "owner_tid": None} + lock = threading.Lock() + + def close_once(reason): + with lock: + request_client = holder.get("client") + owner_tid = holder.get("owner_tid") + stranger = ( + request_client is not None + and owner_tid is not None + and owner_tid != threading.get_ident() + ) + if not stranger: + holder["client"] = None + holder["owner_tid"] = None + if request_client is None: + return + if stranger: + agent._abort_request_openai_client(request_client, reason=reason) + else: + agent._close_request_openai_client(request_client, reason=reason) + + def owner_workload(): + # Set client from owner thread. + with lock: + holder["client"] = sentinel + holder["owner_tid"] = threading.get_ident() + # Simulate work being interrupted by a stranger from outside. + nonlocal_stranger_event.wait(timeout=2.0) + # Worker unwinds — its finally calls close once. + close_once("request_complete") + + nonlocal_stranger_event = threading.Event() + owner = threading.Thread(target=owner_workload) + owner.start() + + # Test thread plays the stranger. + # Give the owner a moment to set the holder. + import time as _t + _t.sleep(0.05) + close_once("interrupt_abort") + nonlocal_stranger_event.set() + owner.join(timeout=5.0) + + assert not owner.is_alive(), "owner thread hung past join timeout" + + # The fix's intended outcome: abort once, close once, holder empty. + assert agent._abort_request_openai_client.call_count == 1 + assert agent._close_request_openai_client.call_count == 1 + assert holder["client"] is None + assert holder["owner_tid"] is None + + +# --------------------------------------------------------------------------- +# End-to-end: the agent's ``_abort_request_openai_client`` shuts sockets and +# logs deferred_close=stranger_thread without ever calling client.close(). +# --------------------------------------------------------------------------- + + +def test_agent_abort_request_openai_client_does_not_call_client_close(caplog): + """``_abort_request_openai_client`` must shutdown sockets but NEVER close(). + + This is the actual entry point used by the stranger-thread path. If a + future refactor accidentally wires it back to ``_close_openai_client`` + the FD race is back. Pin both the shutdown side-effect AND the absence + of any ``client.close()`` call. + """ + from run_agent import AIAgent + + sock = _FakeSocket() + client = _build_fake_client(sock) + + # ``client.close()`` would mutate the holder if invoked — give it a + # MagicMock spy so we can assert no call. + client.close = MagicMock() + + agent = AIAgent.__new__(AIAgent) + agent._client_log_context = lambda: "provider=test" + + with caplog.at_level(logging.INFO, logger="run_agent"): + agent._abort_request_openai_client(client, reason="interrupt_abort") + + # Sockets shut down (one in our fake pool). + assert sock.shutdown_calls == 1 + assert sock.close_calls == 0 + # And critically: client.close() never ran here. + client.close.assert_not_called() + + # The log line is parseable: same ``tcp_force_closed=N`` field shape as + # the existing ``close`` log so dashboards keep working, plus a + # ``deferred_close=stranger_thread`` marker to make the new path + # observable in production triage. + msgs = [r.getMessage() for r in caplog.records] + assert any( + "OpenAI client aborted (interrupt_abort" in m + and "tcp_force_closed=1" in m + and "deferred_close=stranger_thread" in m + for m in msgs + ), f"missing abort log line; got: {msgs!r}" + + +def test_agent_abort_request_openai_client_null_client_is_noop(): + """A ``None`` client must short-circuit cleanly (defensive).""" + from run_agent import AIAgent + + agent = AIAgent.__new__(AIAgent) + agent._client_log_context = lambda: "provider=test" + + # No exception, no side effect. + agent._abort_request_openai_client(None, reason="interrupt_abort") + + +# --------------------------------------------------------------------------- +# FD-recycling proof: when shutdown-only is honored, a stranger-thread abort +# CANNOT release an FD that the owning thread still references. +# --------------------------------------------------------------------------- + + +def test_fd_recycle_window_closed_by_shutdown_only(): + """Construct the exact race the reporter saw — abort from a stranger + thread, then have the (simulated) kernel recycle the FD into a new file. + With the fix, the worker's surviving socket reference cannot be + confused with the recycled file descriptor. + """ + from agent.agent_runtime_helpers import force_close_tcp_sockets + + # Tracks "was the FD released by the abort path?" — that is the only + # signal the kernel needs to recycle the integer to a new ``open()``. + fd_released = {"yes": False} + + class _OwnedSocket: + """Simulates a socket whose FD is shared with the owner's SSL BIO. + + ``close`` flips ``fd_released`` so the test can assert that with + the fix the abort path NEVER releases the FD (and therefore the + kernel never recycles it under the owner's still-active reference). + """ + + def __init__(self): + self.shutdowns = 0 + + def shutdown(self, _how): + self.shutdowns += 1 + + def close(self): + fd_released["yes"] = True + + sock = _OwnedSocket() + client = _build_fake_client(sock) + + # Stranger thread runs the abort sweep (== what asyncio_0 did in the + # reporter's session). + _call_inside_owner_thread(lambda: force_close_tcp_sockets(client)) + + assert sock.shutdowns == 1, "shutdown must wake the worker" + assert fd_released["yes"] is False, ( + "force_close_tcp_sockets released the FD from a stranger thread — " + "this is exactly the #29507 race. The owner thread must own close()." + ) ``` - **#67238: fix(agent): request-local Anthropic clients so the stale/interrupt watchdog never corrupts SQLite (#67142, supersedes #51688)**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `agent/chat_completion_helpers.py`, `run_agent.py`, `tests/agent/test_cascading_interrupt_6600.py`, `tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py`, `tests/run_agent/test_run_agent.py`, `tests/run_agent/test_stream_stale_circuit_breaker.py`, `tests/run_agent/test_streaming.py` - fix hunks: - `agent/chat_completion_helpers.py` ```diff @@ -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") ``` ```diff @@ -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 ``` ```diff @@ -458,7 +465,11 @@ def _abort_active_request(reason: str) -> None: 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 ``` ```diff @@ -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 ``` ```diff @@ -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()``. ``` ```diff @@ -564,24 +580,34 @@ def _close_request_client_once(reason: str) -> None: 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: ``` ```diff @@ -893,11 +919,10 @@ def _call(): 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 ``` ```diff @@ -933,13 +958,12 @@ def _call(): ) # 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") ``` ```diff @@ -2393,6 +2417,10 @@ def _on_reasoning(text): _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- ``` ```diff @@ -2403,9 +2431,10 @@ def _on_reasoning(text): # 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 ``` ```diff @@ -2428,7 +2457,13 @@ def _close_request_client_once(reason: str) -> None: 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) ``` ```diff @@ -2976,13 +3011,18 @@ def _call_chat_completions(stream_attempt_id: int): 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 ``` ```diff @@ -3011,7 +3051,7 @@ def _call_anthropic(): 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 ``` ```diff @@ -3149,8 +3189,16 @@ def _call(): 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 ``` ```diff @@ -3276,13 +3324,11 @@ def _call(): ) _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" ``` ```diff @@ -3341,15 +3387,13 @@ def _call(): # 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" ``` ```diff @@ -3589,11 +3633,14 @@ def _call(): # 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") ``` ```diff @@ -3622,11 +3669,10 @@ def _call(): ) 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") ``` - `run_agent.py` ```diff @@ -4259,6 +4259,103 @@ def _abort_request_openai_client(self, client: Any, *, reason: str) -> None: 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 ``` ```diff @@ -4644,15 +4741,18 @@ def _credential_pool_may_recover_rate_limit(self) -> bool: 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)), ``` - `tests/agent/test_cascading_interrupt_6600.py` ```diff @@ -132,3 +132,93 @@ def _create_1(**kwargs): 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" + ) ``` - `tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py` ```diff @@ -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. - -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. +"""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). + +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 ``` ```diff @@ -41,6 +49,9 @@ def _make_anthropic_agent(**kwargs): agent.api_mode = "anthropic_messages" agent._anthropic_client = MagicMock() agent._anthropic_api_key = "[redacted:secret-assignment]" + # #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 ``` ```diff @@ -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] ``` ```diff @@ -100,14 +114,19 @@ def _stream_side_effect(*args, **kwargs): 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() ``` ```diff @@ -117,14 +136,15 @@ def test_stale_stream_calls_anthropic_rebuild_not_openai(self, monkeypatch): 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()) ``` ```diff @@ -135,8 +155,10 @@ def _blocking_gen(): 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( ``` ```diff @@ -145,6 +167,6 @@ def _blocking_gen(): 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 ``` - `tests/run_agent/test_run_agent.py` ```diff @@ -7648,13 +7648,60 @@ def test_interruptible_has_anthropic_branch(self): 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="[redacted:secret-assignment]", + 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.""" ``` - `tests/run_agent/test_stream_stale_circuit_breaker.py` ```diff @@ -39,6 +39,9 @@ def _make_anthropic_agent(**kwargs): agent.api_mode = "anthropic_messages" agent._anthropic_client = MagicMock() agent._anthropic_api_key = "[redacted:secret-assignment]" + # #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 ``` ```diff @@ -116,7 +119,10 @@ def _stream_side_effect(*args, **kwargs): # 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({}) ``` - `tests/run_agent/test_streaming.py` ```diff @@ -1196,6 +1196,9 @@ def test_anthropic_stream_refreshes_activity_on_every_event(self): 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({}) ``` ```diff @@ -1251,16 +1254,18 @@ def __iter__(self): _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( ``` ```diff @@ -1286,6 +1291,7 @@ def test_generic_anthropic_valueerror_still_propagates_without_stream_retry( 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({}) ``` ```diff @@ -1326,15 +1332,17 @@ def test_anthropic_zero_event_stream_retried_as_transient( 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") ``` ```diff @@ -1369,13 +1377,14 @@ def test_anthropic_eventless_sdk_assertion_normalized_to_empty_stream( 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: ``` - **#29544: fix(api-call): close FD-recycling race that wrote TLS bytes into kanban.db (#29507)**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `5b6f0b695`, `#30858` - **#44066: fix(memory): close SQLite in holographic shutdown + header validation (#44037)**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `7c0a5def58fbec985ace8c887ca3484f7b8cf584`, `#54133`, `#44037`, `#54133` - **#67210: fix(anthropic): abort watchdog TLS sockets from a stranger thread, close from the worker (#67142)**. closed without merge - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#67238`, `#67142`, `#67238`, `#51688` - **#67270: fix(agent): prevent Anthropic stale-stream SQLite corruption via _abort_anthropic_client (#67142)**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `e53f87fe6925f9b0575f16b814d8994dfd455128`, `#67238`, `#67142` ## Fix status - Merged to main in PR #30858 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-07-19`. - Subsystem: `agent`