From 19fb03a93c3eee9d35b87fc9acd50be2924cfc66 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 6 Aug 2026 18:40:09 -0700 Subject: [PATCH] Port from QwenLM/qwen-code#8602: cap a streaming response's total lifetime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stale-stream detector only bounds the gap BETWEEN chunks and resets on every chunk, so a drip-fed stream — a gateway trickling keep-alive-shaped chunks, or a model crawling through one runaway generation for hours — defeats it indefinitely: the turn never completes and the session sits silent until an outer timeout (if any) kills it. Adds a total wall-clock lifetime cap for one streaming response attempt: - agent.stream_max_lifetime (config.yaml) / HERMES_STREAM_MAX_LIFETIME, default 1800s, 0 disables. Never fires before the effective stale-stream timeout, so it cannot preempt reasoning-model patience floors. - Main OpenAI/Anthropic poll loop: tripping the cap kills the connection exactly like a stale kill (attempt cancelled, request client closed), counts in the #58962 cross-turn stale-streak breaker, and lets the bounded retry loop / partial-stub continuation recover. - Bedrock poll loop (sibling site): same cap wired into the existing event watchdog, surfacing a distinct TimeoutError. Tests: drip-fed stream (events flowing every 50ms so the stale detector can never fire) is killed at the cap and bumps the streak — verified to hang without the fix (sabotage run timed out); 0-disable; config/env resolution precedence. Docs: configuration.md timeout table + env var reference. --- agent/chat_completion_helpers.py | 134 +++++++++++++-- hermes_cli/config_defaults.py | 11 ++ tests/run_agent/test_stream_max_lifetime.py | 162 ++++++++++++++++++ .../docs/reference/environment-variables.md | 1 + website/docs/user-guide/configuration.md | 3 + 5 files changed, 300 insertions(+), 11 deletions(-) create mode 100644 tests/run_agent/test_stream_max_lifetime.py diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 0fd556f8403d0..f89e38f86e03d 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -383,6 +383,32 @@ def _derive_stream_stale_timeout(agent, api_kwargs: dict) -> float: return _timeout +def _resolve_stream_max_lifetime(agent) -> float: + """Total wall-clock cap (seconds) for one streaming response attempt. + + Resolution: config.yaml ``agent.stream_max_lifetime`` → env + ``HERMES_STREAM_MAX_LIFETIME`` → default 1800 (30 minutes). ``0`` + disables the cap. Ported from QwenLM/qwen-code#8602 + (``streamMaxLifetimeMs``): the stale detector resets on every chunk, so + only a lifetime cap bounds a drip-fed stream that never goes quiet but + also never completes. + """ + _default = 1800.0 + try: + from hermes_cli.config import load_config_readonly + + _cfg = load_config_readonly() # read-only consumer — no deepcopy + _agent_cfg = _cfg.get("agent") if isinstance(_cfg, dict) else None + if isinstance(_agent_cfg, dict): + _v = _agent_cfg.get("stream_max_lifetime") + if isinstance(_v, (int, float)) and not isinstance(_v, bool): + _default = float(_v) + except Exception: + pass + _lifetime = env_float("HERMES_STREAM_MAX_LIFETIME", _default) + return max(0.0, _lifetime) + + def _bedrock_reasoning_stale_floor(model_id: object) -> "float | None": """Map a Bedrock inference-profile id to its reasoning stale-timeout floor. @@ -2583,6 +2609,12 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= _bedrock_region = api_kwargs.get("__bedrock_region__", "us-east-1") # Same patience budget as the OpenAI/Anthropic stale detector. _bedrock_stale_timeout = _derive_stream_stale_timeout(agent, api_kwargs) + # Total-lifetime cap — same drip-fed-stream class as the main poll + # loop below (the event watchdog resets on every yielded event). + _bedrock_max_lifetime = _resolve_stream_max_lifetime(agent) + if _bedrock_max_lifetime > 0: + _bedrock_max_lifetime = max(_bedrock_max_lifetime, _bedrock_stale_timeout) + _bedrock_lifetime_start = time.time() # Cross-turn stale-stream circuit breaker (#58962): a pre-elevated # streak from prior wedged turns aborts before we even start — mirrors @@ -2726,17 +2758,37 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= # no data, or a silently hung provider). Without this the worker # blocks in ``for event in event_stream`` indefinitely. _stale_elapsed = time.time() - _bedrock_last_event["t"] - if _stale_elapsed > _bedrock_stale_timeout: - logger.warning( - "Bedrock stream stale for %.0fs (threshold %.0fs) — no events " - "received. region=%s model=%s. Aborting call.", - _stale_elapsed, _bedrock_stale_timeout, - _bedrock_region, api_kwargs.get("modelId", "unknown"), - ) - agent._buffer_status( - f"⚠️ No events from Bedrock for {int(_stale_elapsed)}s " - f"(model: {api_kwargs.get('modelId', 'unknown')}). Aborting..." - ) + _bedrock_lifetime_hit = ( + _bedrock_max_lifetime > 0 + and (time.time() - _bedrock_lifetime_start) > _bedrock_max_lifetime + ) + if _stale_elapsed > _bedrock_stale_timeout or _bedrock_lifetime_hit: + if _bedrock_lifetime_hit: + _bedrock_elapsed_total = time.time() - _bedrock_lifetime_start + logger.warning( + "Bedrock stream exceeded max lifetime %.0fs (elapsed " + "%.0fs) — aborting runaway stream. region=%s model=%s.", + _bedrock_max_lifetime, _bedrock_elapsed_total, + _bedrock_region, api_kwargs.get("modelId", "unknown"), + ) + agent._buffer_status( + f"⚠️ Bedrock stream still running after " + f"{int(_bedrock_elapsed_total)}s (cap " + f"{int(_bedrock_max_lifetime)}s). Aborting..." + ) + # Fresh cap in case the (unabortable) worker survives. + _bedrock_lifetime_start = time.time() + else: + logger.warning( + "Bedrock stream stale for %.0fs (threshold %.0fs) — no events " + "received. region=%s model=%s. Aborting call.", + _stale_elapsed, _bedrock_stale_timeout, + _bedrock_region, api_kwargs.get("modelId", "unknown"), + ) + agent._buffer_status( + f"⚠️ No events from Bedrock for {int(_stale_elapsed)}s " + f"(model: {api_kwargs.get('modelId', 'unknown')}). Aborting..." + ) # Count the stale kill in the SAME cross-turn breaker as the # OpenAI/Anthropic path (#58962). _bump_stale_streak(agent) @@ -2766,6 +2818,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= # and the streak carries forward. Break rather than keep polling # a worker we cannot abort. result["error"] = TimeoutError( + f"Bedrock stream exceeded max lifetime " + f"{int(_bedrock_max_lifetime)}s — aborting runaway stream " + f"so the retry/fallback path can recover." + if _bedrock_lifetime_hit else f"Bedrock stream produced no events for {int(_stale_elapsed)}s " f"(threshold {int(_bedrock_stale_timeout)}s) — aborting stalled " f"stream so the retry/fallback path can recover." @@ -4115,6 +4171,27 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= if _reasoning_floor is not None: _stream_stale_timeout = max(_stream_stale_timeout, _reasoning_floor) + # Total-lifetime cap for one streaming response (ported from + # QwenLM/qwen-code#8602). The stale detector above only bounds the gap + # BETWEEN chunks and resets on every chunk, so a drip-fed stream — a + # gateway trickling keep-alive-shaped chunks, or a model crawling through + # one oversized message for hours — defeats it indefinitely. This cap is + # charged on wall-clock time since the CURRENT stream attempt began and + # never resets on chunk arrival. Tripping it kills the connection the + # same way a stale kill does: the worker surfaces a transport error, the + # bounded retry loop may reconnect (each retry gets a fresh cap), and a + # turn that already streamed text returns the partial-stub / continuation + # path instead of sitting silent for hours. 0 disables. + _stream_max_lifetime = _resolve_stream_max_lifetime(agent) + if ( + _stream_max_lifetime > 0 + and _stream_stale_timeout is not None + and _stream_stale_timeout != float("inf") + ): + # Never let the lifetime cap preempt the stale detector's patience. + _stream_max_lifetime = max(_stream_max_lifetime, _stream_stale_timeout) + _stream_lifetime_start = {"t": time.time()} + t = threading.Thread(target=_context_thread_target(_call), daemon=True) t.start() _last_heartbeat = time.time() @@ -4122,6 +4199,41 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= while t.is_alive(): t.join(timeout=0.3) + # Lifetime cap: fires even while chunks are still flowing (the stale + # detector below can never fire in that case). + if _stream_max_lifetime > 0: + _lifetime_elapsed = time.time() - _stream_lifetime_start["t"] + if _lifetime_elapsed > _stream_max_lifetime: + logger.warning( + "Stream exceeded max lifetime %.0fs (elapsed %.0fs) — " + "killing connection. model=%s. A drip-fed or runaway " + "stream never completes on its own; see " + "agent.stream_max_lifetime / HERMES_STREAM_MAX_LIFETIME.", + _stream_max_lifetime, _lifetime_elapsed, + api_kwargs.get("model", "unknown"), + ) + agent._buffer_status( + f"⚠️ Stream still running after {int(_lifetime_elapsed)}s " + f"(cap {int(_stream_max_lifetime)}s) — aborting runaway " + f"response (model: {api_kwargs.get('model', 'unknown')})." + ) + try: + _cancel_current_stream_attempt("stream_max_lifetime_kill") + _close_request_client_once("stream_max_lifetime_kill") + except Exception: + pass + # Count the kill in the SAME cross-turn breaker as a stale + # kill (#58962) — a provider that repeatedly drip-feeds + # never-ending streams is as wedged as one that goes silent. + _bump_stale_streak(agent) + # Fresh cap for any retry attempt the worker's bounded retry + # loop opens after this forced close. + _stream_lifetime_start["t"] = time.time() + last_chunk_time["t"] = time.time() + agent._touch_activity( + f"stream exceeded max lifetime {int(_stream_max_lifetime)}s, aborted" + ) + # Periodic heartbeat: touch the agent's activity tracker so the # gateway's inactivity monitor knows we're alive while waiting # for stream chunks. Without this, long thinking pauses (e.g. diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 41c493383ba5f..31d4bf9eed70f 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -224,6 +224,17 @@ DEFAULT_CONFIG = { # detector instead of hanging forever. The env var # ``HERMES_LOCAL_STREAM_STALE_TIMEOUT`` overrides for escape-hatch use. "local_stream_stale_timeout": 900, + # Total wall-clock cap in seconds for ONE streaming response attempt + # (ported from QwenLM/qwen-code#8602). The stale-stream detector only + # bounds the gap BETWEEN chunks and resets on every chunk, so a + # drip-fed stream — a gateway trickling keep-alive-shaped chunks, or a + # model crawling through one runaway generation — can defeat it + # indefinitely. This cap never resets on chunk arrival; tripping it + # kills the connection like a stale kill (bounded retry / partial-stub + # continuation recover it). Never fires before the effective + # stale-stream timeout. Env override: HERMES_STREAM_MAX_LIFETIME. + # 0 disables. + "stream_max_lifetime": 1800, # How user-attached images are presented to the main model on each turn. # "auto" — attach natively when the active model reports # supports_vision=True AND the user hasn't explicitly diff --git a/tests/run_agent/test_stream_max_lifetime.py b/tests/run_agent/test_stream_max_lifetime.py new file mode 100644 index 0000000000000..71a05ef43c599 --- /dev/null +++ b/tests/run_agent/test_stream_max_lifetime.py @@ -0,0 +1,162 @@ +"""Total-lifetime cap for streaming responses (port of QwenLM/qwen-code#8602). + +The stale-stream detector only bounds the gap BETWEEN chunks and resets on +every chunk, so a drip-fed stream — a gateway trickling keep-alive-shaped +chunks, or a model crawling through one runaway generation — defeats it +indefinitely. These tests cover the lifetime cap added to +``interruptible_streaming_api_call``: + +- a stream that keeps yielding events but never completes is killed once the + cap elapses, even though the stale detector never fires; +- the cap counts the kill in the same cross-turn stale-streak breaker; +- ``0`` disables the cap (a healthy stream completes normally); +- config/env resolution: config.yaml ``agent.stream_max_lifetime`` is the + default, ``HERMES_STREAM_MAX_LIFETIME`` overrides it. + +The harness mirrors tests/run_agent/test_stream_stale_circuit_breaker.py. +""" + +import threading +import time + +import pytest +from unittest.mock import MagicMock + +from types import SimpleNamespace + + +def _make_anthropic_agent(**kwargs): + from run_agent import AIAgent + + defaults = dict( + api_key="test-key", + base_url="https://example.com/v1", + model="claude-opus-4-7", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + defaults.update(kwargs) + agent = AIAgent(**defaults) + agent.api_mode = "anthropic_messages" + agent._anthropic_client = MagicMock() + agent._anthropic_api_key = "test-anthropic-key" + agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client + return agent + + +def _good_stream_cm(): + """Context manager whose stream yields no events and returns a valid message.""" + cm = MagicMock() + stream = MagicMock() + stream.__iter__ = MagicMock(return_value=iter([])) + msg = MagicMock() + msg.content = [] + msg.stop_reason = "end_turn" + msg.usage = SimpleNamespace(input_tokens=10, output_tokens=5) + stream.get_final_message = MagicMock(return_value=msg) + cm.__enter__ = MagicMock(return_value=stream) + cm.__exit__ = MagicMock(return_value=False) + return cm + + +class TestStreamMaxLifetime: + @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning") + def test_drip_fed_stream_killed_at_lifetime_cap(self, monkeypatch): + """A stream that keeps yielding events (so the stale detector never + fires) must still be killed once the total-lifetime cap elapses.""" + # Stale detector armed but never tripped: events arrive every 50ms. + monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "5") + monkeypatch.setenv("HERMES_STREAM_MAX_LIFETIME", "5") + monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "50") + + agent = _make_anthropic_agent() + agent._consecutive_stale_streams = 0 + killed = threading.Event() + + def _drip_gen(): + # Keep-alive-shaped events forever, until the watchdog aborts us. + while not killed.wait(timeout=0.05): + ev = MagicMock() + ev.type = "ping" + yield ev + raise ConnectionError("socket aborted by lifetime watchdog") + + def _stream_side_effect(*args, **kwargs): + cm = MagicMock() + stream = MagicMock() + stream.__iter__ = MagicMock(return_value=_drip_gen()) + cm.__enter__ = MagicMock(return_value=stream) + cm.__exit__ = MagicMock(return_value=False) + return cm + + agent._anthropic_client.messages.stream.side_effect = _stream_side_effect + # The lifetime watchdog aborts the request-local client's sockets from + # the poll thread; simulate the socket shutdown waking the read. + agent._abort_request_anthropic_client = lambda *a, **k: killed.set() + + start = time.time() + with pytest.raises(Exception): + agent._interruptible_streaming_api_call({}) + elapsed = time.time() - start + + # Killed by the cap — far sooner than any stale-detector trip could + # have fired against a stream that never went quiet, and the kill is + # counted in the cross-turn breaker. + assert elapsed < 60 + assert agent._consecutive_stale_streams >= 1 + + @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning") + def test_zero_disables_cap(self, monkeypatch): + """stream_max_lifetime=0 disables the cap; a healthy stream completes.""" + monkeypatch.setenv("HERMES_STREAM_MAX_LIFETIME", "0") + + agent = _make_anthropic_agent() + agent._anthropic_client.messages.stream.return_value = _good_stream_cm() + + resp = agent._interruptible_streaming_api_call({}) + assert resp is not None + assert agent._consecutive_stale_streams == 0 + + +class TestResolveStreamMaxLifetime: + def test_env_overrides_default(self, monkeypatch): + from agent.chat_completion_helpers import _resolve_stream_max_lifetime + + monkeypatch.setenv("HERMES_STREAM_MAX_LIFETIME", "42") + assert _resolve_stream_max_lifetime(MagicMock()) == 42.0 + + def test_config_value_is_default(self, monkeypatch): + import agent.chat_completion_helpers as cch + + monkeypatch.delenv("HERMES_STREAM_MAX_LIFETIME", raising=False) + monkeypatch.setattr( + "hermes_cli.config.load_config_readonly", + lambda: {"agent": {"stream_max_lifetime": 777}}, + ) + assert cch._resolve_stream_max_lifetime(MagicMock()) == 777.0 + + def test_default_when_unset(self, monkeypatch): + import agent.chat_completion_helpers as cch + + monkeypatch.delenv("HERMES_STREAM_MAX_LIFETIME", raising=False) + monkeypatch.setattr( + "hermes_cli.config.load_config_readonly", lambda: {"agent": {}} + ) + assert cch._resolve_stream_max_lifetime(MagicMock()) == 1800.0 + + def test_negative_clamped_to_zero(self, monkeypatch): + from agent.chat_completion_helpers import _resolve_stream_max_lifetime + + monkeypatch.setenv("HERMES_STREAM_MAX_LIFETIME", "-5") + assert _resolve_stream_max_lifetime(MagicMock()) == 0.0 + + def test_bool_config_value_ignored(self, monkeypatch): + import agent.chat_completion_helpers as cch + + monkeypatch.delenv("HERMES_STREAM_MAX_LIFETIME", raising=False) + monkeypatch.setattr( + "hermes_cli.config.load_config_readonly", + lambda: {"agent": {"stream_max_lifetime": True}}, + ) + assert cch._resolve_stream_max_lifetime(MagicMock()) == 1800.0 diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index bae4a6166c20a..d2f615a2cb3c4 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -802,6 +802,7 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us | `HERMES_STREAM_READ_TIMEOUT` | Streaming socket read timeout in seconds (default: `120`). Auto-increased to `HERMES_API_TIMEOUT` for local providers. Increase if local LLMs time out during long code generation. | | `HERMES_STREAM_STALE_TIMEOUT` | Stale stream detection timeout in seconds (default: `180`). Auto-disabled for local providers. Triggers connection kill if no chunks arrive within this window. | | `HERMES_LOCAL_STREAM_STALE_TIMEOUT` | Stale stream ceiling for local providers (Ollama, oMLX, llama-cpp) in seconds (default: `900`). When the base stale timeout is at its default and a local endpoint is detected, this finite ceiling replaces the former infinite disable so a wedged local server eventually trips the detector instead of hanging forever. Also configurable via `agent.local_stream_stale_timeout` in `config.yaml`. | +| `HERMES_STREAM_MAX_LIFETIME` | Total wall-clock cap in seconds for one streaming response attempt (default: `1800`). Unlike the stale-stream detector, this never resets on chunk arrival, so it bounds drip-fed streams that keep sending data but never complete. `0` disables; never fires before the effective stale-stream timeout. Also configurable via `agent.stream_max_lifetime` in `config.yaml`. | | `HERMES_STREAM_RETRIES` | Number of mid-stream reconnect attempts on transient network errors (default: `3`). | | `HERMES_STREAM_STALE_GIVEUP` | Cross-turn circuit breaker: after this many consecutive stale kills (streaming or non-streaming) with no completed response, abort each call immediately with an actionable error instead of re-waiting out the stale timeout (default: `5`, `0` disables). Resets on any completed response, `/model` switch, fallback activation, or turn-start primary restore. | | `HERMES_AGENT_TIMEOUT` | Gateway inactivity timeout for a running agent in seconds (default: `1800`, 30 minutes). Resets on every tool call and streamed token. Set to `0` to disable. | diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index b616fba263a32..04333fbd41ba9 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -973,6 +973,7 @@ Hermes has separate timeout layers for streaming, plus a stale detector for non- |---------|---------|----------------|--------------| | Socket read timeout | 120s | Auto-raised to 1800s | `HERMES_STREAM_READ_TIMEOUT` | | Stale stream detection | 180s | Raised to a 900s ceiling (`agent.local_stream_stale_timeout`) | `HERMES_STREAM_STALE_TIMEOUT` | +| Stream max lifetime | 1800s | Unchanged | `agent.stream_max_lifetime` or `HERMES_STREAM_MAX_LIFETIME` | | Stale non-stream detection | 90s | Auto-disabled when left implicit | `providers..stale_timeout_seconds` or `HERMES_API_CALL_STALE_TIMEOUT` | | API call (non-streaming) | 1800s | Unchanged | `providers..request_timeout_seconds` / `timeout_seconds` or `HERMES_API_TIMEOUT` | @@ -980,6 +981,8 @@ The **socket read timeout** controls how long httpx waits for the next chunk of The **stale stream detection** kills connections that receive SSE keep-alive pings but no actual content. For local providers (which don't send keep-alive pings during prefill) the default is raised to a finite 900-second ceiling instead of the 180s base — configurable via `agent.local_stream_stale_timeout` or the `HERMES_LOCAL_STREAM_STALE_TIMEOUT` env var. +The **stream max lifetime** caps the total wall-clock time of one streaming response. The stale detector resets on every chunk, so a drip-fed stream — a gateway trickling keep-alive-shaped chunks, or a model crawling through one runaway generation — can defeat it indefinitely. The lifetime cap never resets on chunk arrival; when it trips, the connection is killed the same way a stale kill is (bounded retry or partial-response continuation recover it). It never fires before the effective stale-stream timeout, and `0` disables it. Configure via `agent.stream_max_lifetime` in config.yaml. + The **stale non-stream detection** kills non-streaming calls that produce no response for too long. By default Hermes disables this on local endpoints to avoid false positives during long prefills. If you explicitly set `providers..stale_timeout_seconds`, `providers..models..stale_timeout_seconds`, or `HERMES_API_CALL_STALE_TIMEOUT`, that explicit value is honored even on local endpoints. ## Context Pressure Warnings