From 1b897fcaf58e694860b98e98f9b678f3223d07de Mon Sep 17 00:00:00 2001 From: kangyi Date: Fri, 12 Jun 2026 22:55:58 +0900 Subject: [PATCH] fix(streaming): hard ceiling breaks local-provider stream deadlock that wedges workers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Local providers (LM Studio / oMLX / Ollama / llama-cpp) disable the stale-stream detector (timeout=inf) so slow prefill on large contexts is never killed. But that also removes the only mechanism that breaks a half-dead connection — server goes quiet, no SSE terminator, socket left open. The _call thread blocks forever in httpcore read(); the outer 'while t.is_alive(): t.join(0.3)' loop waits on it forever; the worker process can never exit. A worker whose kanban task is already 'done' then lingers in S (sleeping) indefinitely, holding one inference slot open. With PARALLEL=1 on a local GPU each stuck worker shows as '+N queued' on the server, and reap_worker_zombies() (waitpid/WNOHANG) cannot reclaim it — the process never exits, so it is S, not Z. Confirmed via py-spy on a stuck worker (task done, 20+ min in S): Thread-2: join() at chat_completion_helpers.py Thread-35: read() at httpcore/_backends/sync.py Fix: add an absolute hard ceiling (HERMES_STREAM_HARD_TIMEOUT, default 600s) that force-closes the client even when the stale detector is disabled (inf). Healthy prefill keeps delivering chunks and refreshes the timer, so legitimate slow prefill is never killed; only a genuinely dead connection (nothing for 10 min) is reaped, unblocking read() and letting the worker exit. Set the var to 0 to opt out. Adds a regression test that hangs (worker-never-exits) without the fix. --- agent/chat_completion_helpers.py | 47 +++++++ ...test_local_provider_stream_hard_timeout.py | 124 ++++++++++++++++++ 2 files changed, 171 insertions(+) create mode 100644 tests/run_agent/test_local_provider_stream_hard_timeout.py diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 0785347d2c993..9a0f458966b31 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -2301,6 +2301,24 @@ def _call(): else: _stream_stale_timeout = _stream_stale_timeout_base + # Absolute hard ceiling, INDEPENDENT of _stream_stale_timeout. Local + # providers disable the stale detector (inf) so slow prefill is never + # killed — but that also leaves a half-dead connection (server quiet, no + # terminator, socket open) able to wedge the _call thread in read() + # forever. The outer t.join() loop then never returns and the worker + # process can never exit (confirmed via py-spy: Thread blocked in + # httpcore read(), joined-on by the run loop). This ceiling force-closes + # the client after a long no-chunk gap so the deadlock is broken. It does + # NOT regress slow prefill: healthy prefill delivers chunks that refresh + # last_chunk_time, so only a genuinely dead connection is reaped. + try: + _stream_hard_timeout = float( + os.getenv("HERMES_STREAM_HARD_TIMEOUT", 600.0)) + except (TypeError, ValueError): + _stream_hard_timeout = 600.0 + if _stream_hard_timeout <= 0: + _stream_hard_timeout = float("inf") # 0 → opt out + t = threading.Thread(target=_call, daemon=True) t.start() _last_heartbeat = time.time() @@ -2328,6 +2346,35 @@ def _call(): # but delivering no real chunks. Kill the client so the # inner retry loop can start a fresh connection. _stale_elapsed = time.time() - last_chunk_time["t"] + + # Hard ceiling (independent of _stream_stale_timeout): break the + # deadlock when a half-dead connection delivers nothing for far longer + # than any legitimate prefill. Without this, local providers (stale + # timeout = inf) wedge the worker permanently — it never exits and + # holds an inference slot open. Force-close exactly like the stale + # path; the blocked read() unwinds with EOF and the join() returns. + if ( + _stale_elapsed > _stream_hard_timeout + and _stream_hard_timeout != float("inf") + ): + logger.warning( + "Stream hard timeout: no chunks for %.0fs (ceiling %.0fs, " + "stale_timeout=%s). model=%s — force-closing to avoid a " + "wedged worker.", + _stale_elapsed, _stream_hard_timeout, + _stream_stale_timeout, api_kwargs.get("model", "unknown"), + ) + try: + _close_request_client_once("stream_hard_timeout_kill") + except Exception: + pass + try: + agent._replace_primary_openai_client( + reason="stream_hard_timeout_pool_cleanup") + except Exception: + pass + last_chunk_time["t"] = time.time() + if _stale_elapsed > _stream_stale_timeout: _est_ctx = estimate_request_context_tokens(api_kwargs) logger.warning( diff --git a/tests/run_agent/test_local_provider_stream_hard_timeout.py b/tests/run_agent/test_local_provider_stream_hard_timeout.py new file mode 100644 index 0000000000000..78945002b7d8b --- /dev/null +++ b/tests/run_agent/test_local_provider_stream_hard_timeout.py @@ -0,0 +1,124 @@ +"""Regression test for the local-provider streaming deadlock on worker exit. + +Bug: when the model endpoint is a local provider (LM Studio / oMLX / Ollama / +llama-cpp), ``_interruptible_streaming_api_call`` sets the stale-stream +timeout to ``float("inf")`` so that slow prefill on large contexts is never +killed. But that also removes the only mechanism that breaks a *half-dead* +connection — one where the server has gone quiet (no terminator, socket still +open). The ``_call`` thread blocks forever in ``httpcore ... read()``, the +outer ``while t.is_alive(): t.join(0.3)`` loop waits on it forever, and the +worker process never exits. With ``PARALLEL=1`` on a local GPU each stuck +worker holds an inference slot, producing ``+N queued`` on the server. + +Confirmed in production via py-spy: a worker whose kanban task was already +``done`` sat in ``S`` (sleeping) for 20+ minutes with: + Thread-2: join() at chat_completion_helpers.py:2487 + Thread-35: read() at httpcore/_backends/sync.py:128 + +Fix: an absolute hard ceiling (``HERMES_STREAM_HARD_TIMEOUT``, default 600s) +that force-closes the client even when the stale detector is disabled (inf), +breaking the deadlock so the worker can exit. Healthy prefill keeps delivering +chunks and refreshes the timer, so it is never killed. +""" +from __future__ import annotations + +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + + +class _DeadStream: + """An SSE stream that accepts the connection but never yields a chunk — + mimics a local server that goes quiet without closing the socket. + + A real socket ``read()`` would block forever; the hard-ceiling fix + force-closes the underlying client, which severs that socket and makes + ``read()`` raise. We emulate that here: ``__next__`` short-polls a + ``closed`` event that the fix sets (via the patched force-close hook). + Once set, the stream ends (severed connection → StopIteration).""" + + def __init__(self, closed_event: threading.Event): + self._closed = closed_event + + def __iter__(self): + return self + + def __next__(self): + # Short poll so that once the force-close fires, the next iteration + # promptly unwinds the _call thread (mirrors read() raising on a + # severed socket). Hard cap keeps a buggy build from hanging the test. + for _ in range(200): # ~20s absolute cap + if self._closed.wait(timeout=0.1): + raise StopIteration + raise StopIteration + + +def _make_local_agent(): + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="http://localhost:1234/v1", # local provider → stale timeout = inf + model="qwen/qwen3-coder-next", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "chat_completions" + agent._interrupt_requested = False + return agent + + +@patch("run_agent.AIAgent._replace_primary_openai_client") +@patch("run_agent.AIAgent._create_request_openai_client") +@patch("run_agent.AIAgent._close_request_openai_client") +def test_local_provider_dead_stream_is_force_closed_by_hard_ceiling( + mock_close, mock_create, mock_replace, monkeypatch +): + """A local-provider stream that never delivers a chunk must be force-closed + by the hard ceiling instead of hanging the worker forever.""" + # Small hard ceiling so the test is fast. + monkeypatch.setenv("HERMES_STREAM_HARD_TIMEOUT", "1.0") + + closed_event = threading.Event() + # The hard-ceiling fix force-closes via the request-client teardown and a + # primary-client replacement. Either firing means the deadlock was broken; + # both set the event so the dead stream unwinds (mirrors a severed socket). + mock_close.side_effect = lambda *a, **k: closed_event.set() + mock_replace.side_effect = lambda *a, **k: closed_event.set() + + mock_client = MagicMock() + mock_client.chat.completions.create.return_value = _DeadStream(closed_event) + mock_create.return_value = mock_client + + agent = _make_local_agent() + + done = threading.Event() + result_box = {} + + def _run(): + try: + result_box["ret"] = agent._interruptible_streaming_api_call({}) + except BaseException as exc: # the function may raise on the killed stream + result_box["exc"] = exc + finally: + done.set() + + runner = threading.Thread(target=_run, daemon=True) + runner.start() + + # The hard ceiling is 1.0s; allow generous slack. Without the fix this + # NEVER completes (worker hangs forever) → the test times out / fails. + finished = done.wait(timeout=8.0) + + assert finished, ( + "interruptible_streaming_api_call hung on a dead local-provider stream " + "— the hard ceiling did not force-close the connection. This is the " + "worker-never-exits deadlock." + ) + # The fix breaks the deadlock by force-closing / replacing the client. + assert closed_event.is_set(), ( + "expected the hard ceiling to force-close (or replace) the client" + )