diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index eda41e199e..20c4be4547 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -3010,12 +3010,21 @@ def _iter_httpx_pool_objects(http_client: Any): def _connection_candidates(conn: Any): - """Walk nested ``_connection`` wrappers (proxy tunnel → HTTP11/2).""" + """Walk nested wrappers: proxy tunnels (``_connection``) plus httpx/httpcore + stream envelopes (``_stream``/``_httpcore_stream``: BoundSyncStream → + ResponseStream → connection byte stream → HTTP11/2 connection).""" seen: set[int] = set() - while conn is not None and id(conn) not in seen: - seen.add(id(conn)) - yield conn - conn = getattr(conn, "_connection", None) + stack = [conn] + while stack: + obj = stack.pop() + if obj is None or id(obj) in seen: + continue + seen.add(id(obj)) + yield obj + for attr in ("_connection", "_stream", "_httpcore_stream"): + nxt = getattr(obj, attr, None) + if nxt is not None and id(nxt) not in seen: + stack.append(nxt) def _socket_from_stream(stream: Any): diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 13dc0dfbd1..04ba798309 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -2641,6 +2641,14 @@ class _StreamingCall(StreamingWaitMonitor): # tolerates that, so the raw read timeout must not fire first. read = stale logger.debug("Cloud reasoning stream — read timeout raised to %.0fs to match stale-stream detector", read) + if stale is not None and stale != float("inf") and stale < read: + # The stale detector would kill this stream at ``stale`` anyway; align + # the worker's own read timeout so the WORKER raises first, closes its + # own response, and retries — descriptor release stays on the owner + # thread instead of needing a stranger-thread close to unwedge a + # parked recv (shutdown() does not unblock every platform, #110769). + read = stale + logger.debug("Stale timeout %.0fs below read timeout — read capped to match so the owner unwinds itself", read) return base, read, 30.0 @staticmethod @@ -3096,7 +3104,12 @@ class _StreamingCall(StreamingWaitMonitor): logger.debug("Streaming worker caught %s after request cancellation — exiting without retry.", type(e).__name__) return False _is_timeout = isinstance(e, (_httpx.ReadTimeout, _httpx.ConnectTimeout, _httpx.PoolTimeout)) - _is_conn_err = isinstance(e, (_httpx.ConnectError, _httpx.RemoteProtocolError, ConnectionError)) + # ReadError is the aborted/reset body read (a stale-killed attempt's socket + # was shut down under a parked reader, ECONNRESET mid-body): the retry loop + # owns recovery. Without it an abort-induced read failure ended the turn + # instead of reconnecting — the "no response ... reconnecting" loop with + # no recovery. + _is_conn_err = isinstance(e, (_httpx.ConnectError, _httpx.ReadError, _httpx.RemoteProtocolError, ConnectionError)) _is_stream_parse_err = self.agent._is_provider_stream_parse_error(e) _is_empty_stream = isinstance(e, EmptyStreamError) _is_sse_conn_err = not _is_timeout and not _is_conn_err and _is_sse_connection_error(e) @@ -3202,6 +3215,59 @@ class _StreamingCall(StreamingWaitMonitor): finally: self._call_done.set() + def _shutdown_stale_attempt_socket(self, response: Any) -> None: + """Best-effort ``shutdown()`` on the killed attempt's socket (monitor thread). + + The pool sweep in ``close_once`` can miss a connection that is checked + out for the in-flight body read. ``shutdown(SHUT_RDWR)`` is FD-safe + from any thread — it wakes the owner's ``recv`` without releasing the + descriptor — so the worker unwinds and releases its own response on + the owner thread (``_call``'s ``except``/``finally``). Never + ``close()`` here: releasing a live TLS descriptor from a stranger + thread lets the kernel recycle it under the owner's SSL BIO, which is + exactly what the shutdown-only rule in ``_abort_request_slot_client`` + forbids (it covers request-local clients too, #30858). + """ + if response is None or response is not self._attempt_stream_response: + return + try: + import socket as _socket + from agent.agent_runtime_helpers import _connection_candidates, _socket_from_stream + exts = getattr(response, "extensions", None) or {} + direct = exts.get("network_stream") if isinstance(exts, dict) else None + stream_obj = getattr(response, "stream", None) + inner = getattr(stream_obj, "_httpcore_stream", None) + if inner is None: + # httpx 0.28 nests one level deeper: BoundSyncStream._stream + # (ResponseStream)._httpcore_stream (httpcore byte stream). + inner = getattr(getattr(stream_obj, "_stream", None), "_httpcore_stream", None) + seen: set = set() + for start in (direct, inner, stream_obj, response): + if start is None or id(start) in seen: + continue + seen.add(id(start)) + for candidate in _connection_candidates(start): + net = getattr(candidate, "_network_stream", None) or getattr(candidate, "_stream", None) + sock = _socket_from_stream(net) if net is not None else None + if sock is None: + sock = _socket_from_stream(candidate) + if sock is None: + continue + try: + settimeout = getattr(sock, "settimeout", None) + if callable(settimeout): + with contextlib.suppress(OSError): + settimeout(0) + sock.shutdown(_socket.SHUT_RDWR) + logger.info("Shut down the stale stream's socket to unblock the reader " + "(attempt superseded; model=%s).", self.api_kwargs.get("model", "unknown")) + except OSError: + pass # already shut / not connected: benign + return + logger.debug("Stale stream socket shutdown found no socket; pool sweep is the only abort") + except Exception: + logger.debug("Stale stream socket shutdown failed", exc_info=True) + def _kill_stale_stream(self, elapsed: float) -> None: """SSE pings but no chunks: cancel the attempt and abort the request-local client so the retry loop opens a fresh one. The shared client is never @@ -3216,9 +3282,14 @@ class _StreamingCall(StreamingWaitMonitor): self.agent._buffer_status( f"⚠️ No response from provider for {int(elapsed)}s (model: {self.api_kwargs.get('model', 'unknown')}, " f"context: ~{_est_ctx:,} tokens). Reconnecting...") + # Captured BEFORE the cancel/abort: the pool sweep can miss a checked-out + # connection, so shut down the killed attempt's own socket too — still + # shutdown-only, never close (see the helper). + _killed_response = self._attempt_stream_response with contextlib.suppress(Exception): self._cancel_current_stream_attempt("stale_stream_kill") self.clients.close_once("stale_stream_kill") + self._shutdown_stale_attempt_socket(_killed_response) _bump_stale_streak(self.agent) # circuit breaker, see ``_stale_streak()`` # Reset the timer so we don't kill repeatedly while the worker unwinds. self.last_chunk_time["t"] = time.time() diff --git a/tests/agent/test_stream_stale_kill_unwedge.py b/tests/agent/test_stream_stale_kill_unwedge.py new file mode 100644 index 0000000000..e60bc9cf1f --- /dev/null +++ b/tests/agent/test_stream_stale_kill_unwedge.py @@ -0,0 +1,266 @@ +"""A stale-killed provider stream must actually unwedge its reader. + +Report #110769: at agent-scale context the client logged "No response from +provider for 180-240s ... Reconnecting" over and over and then gave up with +"The model server is not responding". The monitor-side kill aborts the +request client's sockets (``force_close_tcp_sockets`` -> ``shutdown(SHUT_RDWR)``), +and that is best-effort: the pool sweep can miss a connection that is checked +out for the in-flight body read. When the reader stays parked the worker never +unwinds, so the retry loop never retries; the monitor re-kills every stale +interval and the call only ends at the byte-read timeout, far past the stale +budget — exactly the reported reconnect loop. + +The kill therefore also shuts down the killed attempt's own socket (resolved +from the identity-guarded ``_attempt_stream_response``) — ``shutdown()`` is +FD-safe from the monitor thread, while ``close()`` from a stranger thread +would release a live TLS descriptor under the owner's SSL BIO (shutdown-only +rule, #30858). And an abort-induced ``httpx.ReadError`` counts as transient +so the retry actually reconnects. Descriptor release stays on the worker: +``_call``'s ``except``/``finally`` closes the managed stream and the request +client on the owner thread. +""" +import json +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import SimpleNamespace + +import pytest + +import run_agent +from agent import chat_completion_helpers as helpers + + +# ── unit: the kill shuts down the killed attempt's socket, never closes it ── + + +def _agent(): + return run_agent.AIAgent( + api_key="test-key", base_url="http://127.0.0.1:1/v1", model="m", provider="custom", + quiet_mode=True, skip_context_files=True, skip_memory=True, enabled_toolsets=[], max_iterations=1, + ) + + +def _call(agent, model="m"): + call = helpers._StreamingCall(agent, {"model": model, "messages": [{"role": "user", "content": "hi"}]}, None) + call._stream_stale_timeout = 5.0 + return call + + +def _fake_tls_response(sock): + """httpx-shaped response: ``stream._httpcore_stream._connection`` -> socket.""" + def _no_close(): + raise AssertionError("monitor must never close") + conn = SimpleNamespace(_network_stream=SimpleNamespace(_sock=sock), _connection=None) + return SimpleNamespace(close=_no_close, + stream=SimpleNamespace(_httpcore_stream=SimpleNamespace(_connection=conn))) + + +def test_stale_kill_shuts_down_but_never_closes_the_killed_attempts_response(): + import socket as _socket + reader, writer = _socket.socketpair() + try: + call = _call(_agent()) + call._attempt_stream_response = _fake_tls_response(reader) + call._kill_stale_stream(7.0) # must not raise via the AssertionError close + reader.setblocking(False) + try: + data = reader.recv(1) + except (BlockingIOError, OSError): + data = b"" + assert data == b"", "the killed attempt's socket was not shut down" + finally: + reader.close() + writer.close() + + +def test_escalation_ignores_a_response_a_racing_retry_replaced(): + """A retry replaces ``_attempt_stream_response``; touching the *new* one would + disturb a healthy stream, so the escalation is identity-guarded.""" + import socket as _socket + reader, writer = _socket.socketpair() + try: + call = _call(_agent()) + live = _fake_tls_response(reader) + call._attempt_stream_response = live + call._shutdown_stale_attempt_socket(SimpleNamespace(close=lambda: None, stream=None)) + call._shutdown_stale_attempt_socket(None) + assert reader.getsockopt(_socket.SOL_SOCKET, _socket.SO_ERROR) == 0 + # ...and the socket it was handed for the killed attempt really is shut down. + call._shutdown_stale_attempt_socket(live) + reader.setblocking(False) + try: + data = reader.recv(1) + except (BlockingIOError, OSError): + data = b"" + assert data == b"" + finally: + reader.close() + writer.close() + + +def test_stale_kill_survives_a_response_without_socket(): + call = _call(_agent()) + call._attempt_stream_response = SimpleNamespace() # no stream/socket: must not raise + call._kill_stale_stream(7.0) + + +def _recv_shutdown_proof(reader, writer): + """Behavioral proof the helper reached *this* socket: the helper applies + ``settimeout(0)`` then ``shutdown(SHUT_RDWR)`` (which emits FIN), so the + peer must observe EOF. An untouched socket would still be blocking with + its default timeout and the peer recv would time out instead.""" + assert reader.gettimeout() == 0.0, "helper never reached this socket" + writer.settimeout(5) + assert writer.recv(1) == b"", "the killed attempt's socket was not shut down" + + +def test_shutdown_reaches_socket_through_real_httpx_wrapper_shape(): + """httpx 0.28 nests BoundSyncStream._stream(ResponseStream)._httpcore_stream + (PoolByteStream)._stream(HTTP11ConnectionByteStream)._connection + (HTTP11Connection)._network_stream(SyncStream)._sock; a lookup that only + follows ``_connection`` from ``response.stream`` never reaches the socket + (review blocker). Shape verified against a live httpx 0.28.1/httpcore 1.0.9 + loopback probe.""" + import socket as _socket + reader, writer = _socket.socketpair() + try: + def _no_close(): + raise AssertionError("monitor must never close") + # Exact live shape: PoolByteStream carries the byte stream under _stream + # (no _connection); the HTTP/1.1 connection carries the socket under + # _network_stream (no _stream). + pool_stream = SimpleNamespace(_stream=None, _connection=None) + h11_byte_stream = SimpleNamespace(_connection=None) + h11_conn = SimpleNamespace(_network_stream=SimpleNamespace(_sock=reader), + _stream=None, _connection=None, _httpcore_stream=None) + h11_byte_stream._connection = h11_conn + pool_stream._stream = h11_byte_stream + resp = SimpleNamespace( + close=_no_close, + stream=SimpleNamespace(_stream=SimpleNamespace(_httpcore_stream=pool_stream)), + ) + call = _call(_agent()) + call._attempt_stream_response = resp + call._shutdown_stale_attempt_socket(resp) # must not raise via close + _recv_shutdown_proof(reader, writer) + finally: + reader.close() + writer.close() + + +def test_shutdown_prefers_response_extensions_network_stream(): + """``response.extensions["network_stream"]`` is the direct socket path + (reviewer's validated probe); it must work even with no stream wrappers.""" + import socket as _socket + reader, writer = _socket.socketpair() + try: + def _no_close(): + raise AssertionError("monitor must never close") + net = SimpleNamespace(get_extra_info=lambda _name: reader) + resp = SimpleNamespace(close=_no_close, stream=None, + extensions={"network_stream": net}) + call = _call(_agent()) + call._attempt_stream_response = resp + call._shutdown_stale_attempt_socket(resp) + _recv_shutdown_proof(reader, writer) + finally: + reader.close() + writer.close() + + +# ── end to end: a reader parked on a silent provider must reconnect ── + + +class _SilentFirstRequest: + """Request #1: 200 + SSE headers, then NOTHING (a provider that accepted the + connection and never produced a byte). Later requests: a normal completion.""" + + def __init__(self): + self.completions = [] + self.release = threading.Event() + hits = self + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *_a): + pass + + def do_POST(self): + n = int(self.headers.get("content-length", 0)) + body = json.loads(self.rfile.read(n) or b"{}") + if not self.path.endswith("/chat/completions"): + self.send_response(404) + self.send_header("content-length", "0") + self.end_headers() + return + hits.completions.append(body) + first = len(hits.completions) == 1 + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + if first: + # Hold the body open with no bytes at all: the client parks in + # its body read. Bounded wait so teardown can never hang. + hits.release.wait(timeout=30.0) + return + try: + chunk = {"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": "m", + "choices": [{"index": 0, "delta": {"content": "reconnected"}, "finish_reason": None}]} + self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode()) + fin = {"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": "m", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]} + self.wfile.write(b"data: " + json.dumps(fin).encode() + b"\n\n") + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + except OSError: + pass # the client aborted us (expected) + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.server.daemon_threads = True + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.base_url = f"http://127.0.0.1:{self.server.server_address[1]}/v1" + + def close(self): + self.release.set() + self.server.shutdown() + self.server.server_close() + + +@pytest.fixture +def silent_wire(): + wire = _SilentFirstRequest() + yield wire + wire.close() + + +def test_wedged_stream_unwinds_within_its_stale_budget_and_reconnects(silent_wire, monkeypatch): + """Pre-fix the worker stayed parked in the body read: no second request was + ever issued and the call only ended at the byte-read timeout (>= 20s here, + 120s by default). Now the kill unblocks the reader, the abort-induced read + error is transient, and the retry lands.""" + monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "2") + monkeypatch.setenv("HERMES_STREAM_RETRIES", "1") + monkeypatch.setenv("HERMES_STREAM_READ_TIMEOUT", "20") + agent = run_agent.AIAgent( + api_key="test-key", base_url=silent_wire.base_url, model="m", provider="custom", + platform="cli", # worker thread + monitor thread: the gateway shape from the report + quiet_mode=True, skip_context_files=True, skip_memory=True, enabled_toolsets=[], max_iterations=1, + ) + agent.api_mode = "chat_completions" + agent._interrupt_requested = False + + started = time.time() + response = agent._interruptible_streaming_api_call( + {"model": "m", "messages": [{"role": "user", "content": "hi"}]}) + elapsed = time.time() - started + + assert len(silent_wire.completions) >= 2, ( + f"the killed attempt never reconnected (requests={len(silent_wire.completions)}); " + "the reader stayed parked instead of unwinding" + ) + assert response.choices[0].message.content == "reconnected" + # The stale budget is 2s; the byte-read timeout is 20s. A parked reader can + # only be recovered by the read timeout, which is what this pins out. + assert elapsed < 12.0, f"stream took {elapsed:.1f}s to recover — the reader was not unwedged by the kill"