diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index cee392caaba69..28fba7c146282 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -138,6 +138,7 @@ def interruptible_api_call(agent, api_kwargs: dict): """ result = {"response": None, "error": None} request_client_holder = {"client": None, "owner_tid": None} + 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 @@ -149,9 +150,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()``. @@ -185,7 +187,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) @@ -205,7 +213,16 @@ def _call(): on_first_delta=getattr(agent, "_codex_on_first_delta", None), ) elif agent.api_mode == "anthropic_messages": - result["response"] = agent._anthropic_messages_create(api_kwargs) + request_client = _set_request_client( + agent._create_request_anthropic_client( + reason="anthropic_messages_request" + ), + kind="anthropic_messages", + ) + result["response"] = agent._anthropic_messages_create( + api_kwargs, + client=request_client, + ) elif agent.api_mode == "bedrock_converse": # Bedrock uses boto3 directly — no OpenAI client needed. # normalize_converse_response produces an OpenAI-compatible @@ -499,11 +516,7 @@ 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") + _close_request_client_once("stale_call_kill") except Exception: pass agent._touch_activity( @@ -536,13 +549,11 @@ 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. For Anthropic Messages the SDK client is + # shared; closing it here would release TLS FDs from the poll + # thread and can resurrect #29507 (TLS bytes written into SQLite). 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") @@ -1722,6 +1733,7 @@ def _on_reasoning(text): result = {"response": None, "error": None, "partial_tool_names": []} request_client_holder = {"client": None, "diag": None, "owner_tid": None} + 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- @@ -1732,9 +1744,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 @@ -1757,7 +1770,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) @@ -2159,7 +2178,7 @@ def _call_chat_completions(): 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 @@ -2183,7 +2202,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 @@ -2217,7 +2236,7 @@ def _call_anthropic(): pass if agent._interrupt_requested: - break + raise InterruptedError("Agent interrupted during Anthropic stream") event_type = getattr(event, "type", None) @@ -2266,8 +2285,13 @@ 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() + 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() return # success @@ -2631,11 +2655,7 @@ def _call(): "(not a network error)." ) try: - if agent.api_mode == "anthropic_messages": - agent._anthropic_client.close() - agent._rebuild_anthropic_client() - else: - _close_request_client_once("stream_interrupt_abort") + _close_request_client_once("stream_interrupt_abort") except Exception: pass raise InterruptedError("Agent interrupted during streaming API call") diff --git a/run_agent.py b/run_agent.py index 63050980934b2..d624ee41d8245 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3762,6 +3762,86 @@ 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`` remains the long-lived primary, but + stale/interrupt handling must not close it from the poll thread. A + per-request client lets the stranger thread abort sockets while the + owning worker performs the SDK-level close, matching the #29507 close + discipline used by OpenAI-wire requests. + """ + 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: + if client is None: + return + try: + 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. + + As with OpenAI request clients, stranger threads must not call the SDK + ``close()`` method because that can release TLS FDs while the owning + worker's SSL BIO is still unwinding (#29507). + """ + 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 @@ -4101,15 +4181,16 @@ 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): + 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 + request_client = client or self._anthropic_client return create_anthropic_message( - self._anthropic_client, + request_client, api_kwargs, log_prefix=getattr(self, "log_prefix", ""), prefer_stream=not bool(getattr(self, "_disable_streaming", False)), diff --git a/tests/agent/test_cascading_interrupt_6600.py b/tests/agent/test_cascading_interrupt_6600.py index 58fc28c4df0f6..9d32e5049e340 100644 --- a/tests/agent/test_cascading_interrupt_6600.py +++ b/tests/agent/test_cascading_interrupt_6600.py @@ -132,3 +132,149 @@ def _create_1(**kwargs): with pytest.raises(httpx.RemoteProtocolError): cch.interruptible_api_call(agent, {"model": "x", "messages": []}) + + + +def _make_anthropic_agent(): + agent = _make_agent() + agent.api_mode = "anthropic_messages" + agent._anthropic_client = MagicMock() + agent._rebuild_anthropic_client = MagicMock() + agent._anthropic_messages_create = MagicMock() + agent._create_request_anthropic_client = MagicMock() + agent._close_request_anthropic_client = MagicMock() + agent._abort_request_anthropic_client = MagicMock() + return agent + + +def _wait_for_mock_call(mock, timeout=2.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_interrupt_does_not_close_shared_client(): + """#29507: interrupt polling must not close the shared Anthropic SDK client. + + The shared client may be in use on the worker thread. Closing it from the + poll thread can release a TLS FD that SQLite later reuses, letting the SSL + BIO write a TLS record into state.db's header. + """ + agent = _make_anthropic_agent() + request_client = MagicMock() + agent._create_request_anthropic_client.return_value = request_client + + 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.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", + ) + _wait_for_mock_call(agent._close_request_anthropic_client) + agent._close_request_anthropic_client.assert_called_with( + request_client, + reason="request_complete", + ) + + +def test_anthropic_non_streaming_stale_does_not_close_shared_client(monkeypatch): + """#29507: stale-call polling must not close/rebuild shared Anthropic client.""" + agent = _make_anthropic_agent() + request_client = MagicMock() + agent._create_request_anthropic_client.return_value = request_client + agent._compute_non_stream_stale_timeout.return_value = 0.05 + agent._codex_silent_hang_hint = MagicMock(return_value=None) + + def _create(_api_kwargs, *, client): + assert client is request_client + time.sleep(2.5) + return object() + + agent._anthropic_messages_create.side_effect = _create + + with pytest.raises(TimeoutError): + cch.interruptible_api_call(agent, {"model": "x", "messages": []}) + + 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="stale_call_kill", + ) + _wait_for_mock_call(agent._close_request_anthropic_client) + agent._close_request_anthropic_client.assert_called_with( + request_client, + reason="request_complete", + ) + + +def test_anthropic_streaming_interrupt_does_not_close_shared_client(monkeypatch): + """#29507: streaming interrupt must not close/rebuild shared Anthropic client.""" + agent = _make_anthropic_agent() + request_client = MagicMock() + agent._create_request_anthropic_client.return_value = request_client + agent._compute_stream_stale_timeout.return_value = 5.0 + agent._invoke_bedrock_stream = MagicMock() + agent.provider = "anthropic" + + class _Chunk: + type = "content_block_delta" + delta = types.SimpleNamespace(type="text_delta", text="hello") + + class _Stream: + def __init__(self): + self.exited = threading.Event() + self.get_final_message = MagicMock( + side_effect=AssertionError( + "get_final_message() must not run after interrupt" + ) + ) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + self.exited.set() + return False + + def __iter__(self): + agent._interrupt_requested = True + time.sleep(1.0) + yield _Chunk() + + stream = _Stream() + request_client.messages.stream.return_value = stream + + with pytest.raises(InterruptedError): + cch.interruptible_streaming_api_call(agent, {"model": "x", "messages": []}) + + 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="stream_interrupt_abort", + ) + _wait_for_mock_call(agent._close_request_anthropic_client) + agent._close_request_anthropic_client.assert_called_with( + request_client, + reason="stream_request_complete", + ) + assert stream.exited.wait(1.0) + stream.get_final_message.assert_not_called()