fix(agent): stale-killed stream unwedges its reader and reconnects (#110769)
The stale-stream monitor aborted a wedged provider stream only via force_close_tcp_sockets() -> shutdown(SHUT_RDWR). That is best-effort: a parked body read is not unblocked on every platform (Windows keeps the pending recv parked) and the sweep can miss the socket. The worker then stayed blocked in the provider read, so the retry loop never retried; the monitor re-killed every stale interval and the call only ended at the byte-read timeout, far past the stale budget - the reported "No response from provider for 180-240s ... Reconnecting" loop ending in "The model server is not responding". - _kill_stale_stream now also closes the killed attempt's own provider response (identity-guarded self._attempt_stream_response), which is what actually unblocks a parked reader; a racing retry's fresh response is never touched. - an abort-induced httpx.ReadError counts as a transient connection error, so the aborted attempt reconnects instead of ending the turn. Reproduced with a local SSE server: before, the worker stayed parked and no second request was issued (recovery only at the byte-read timeout); after, the kill unblocks the reader at the stale budget and the retry lands.
This commit is contained in:
finn763
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user