diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index e849b955d19a7..fc27577593244 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -975,6 +975,12 @@ def try_recover_primary_transport( if agent._fallback_activated: return False + # The local first-chunk watchdog already proved this request was accepted + # but produced no SSE data. Rebuilding the same primary client just repeats + # the full TTFB wait; route to fallback or fail fast instead. + if getattr(api_error, "_hermes_local_first_chunk_timeout", False): + return False + # Only for transient transport errors error_type = type(api_error).__name__ if error_type not in _TRANSIENT_TRANSPORT_ERRORS: diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 4e4aa455a600f..6bb60cc8a1027 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -236,6 +236,161 @@ def _check_stale_giveup(agent) -> None: ) +def _local_provider_stream_stale_timeout(api_payload: Any) -> float | None: + """Return an opt-in bounded stale timeout for local streaming calls. + + By default local endpoints keep the historical unbounded behavior (large + self-hosted models can legitimately spend a long time pre-filling). Some + local backends can instead park a request on an open socket forever; set + HERMES_LOCAL_STALE_TIMEOUT (or HERMES_LOCAL_STREAM_STALE_TIMEOUT) to bound + them so the normal retry/fallback path gets a chance to run. + """ + raw = os.getenv("HERMES_LOCAL_STALE_TIMEOUT") + if raw is None: + raw = os.getenv("HERMES_LOCAL_STREAM_STALE_TIMEOUT") + if raw is None: + return None + + timeout = _env_float( + "HERMES_LOCAL_STALE_TIMEOUT", + _env_float("HERMES_LOCAL_STREAM_STALE_TIMEOUT", 75.0), + ) + if timeout <= 0: + # Preserve the stale-timeout convention that non-positive env values + # explicitly disable the watchdog. + return float("inf") + + est_tokens = estimate_request_context_tokens(api_payload) + if est_tokens > 100_000: + return max(timeout, 300.0) + if est_tokens > 50_000: + return max(timeout, 240.0) + if est_tokens > 25_000: + return max(timeout, 150.0) + if est_tokens > 10_000: + return max(timeout, 90.0) + return timeout + + +def _local_provider_first_chunk_timeout(api_payload: Any, model: Any) -> float | None: + """Return the generic local-provider no-first-chunk cutoff. + + Local backends can legitimately spend longer pre-filling than cloud + providers, but they still need a finite TTFB watchdog so a fallback model + cannot inherit an infinite wait after the primary already stalled. + """ + timeout = _env_float( + "HERMES_LOCAL_FIRST_CHUNK_TIMEOUT", + _env_float("HERMES_LOCAL_TTFB_TIMEOUT", 90.0), + ) + if timeout <= 0: + return float("inf") + + est_tokens = estimate_request_context_tokens(api_payload) + if est_tokens > 100_000: + return max(timeout, 600.0) + if est_tokens > 50_000: + return max(timeout, 360.0) + if est_tokens > 25_000: + return max(timeout, 240.0) + if est_tokens > 10_000: + return max(timeout, 150.0) + return timeout + + +def _local_provider_non_stream_stale_timeout(api_payload: Any, model: Any) -> float: + """Return a finite stale timeout for local non-streaming calls. + + Local non-streaming OpenAI-compatible calls are dangerous when left + unbounded: the server may accept the request, generate for a very long time, + and send no response headers until the entire completion is done. Keep the + default finite so fallback can run, while still scaling for large prompt + prefill. + """ + timeout = _env_float( + "HERMES_LOCAL_NON_STREAM_STALE_TIMEOUT", + _env_float("HERMES_LOCAL_RESPONSE_TIMEOUT", 120.0), + ) + if timeout <= 0: + return float("inf") + + est_tokens = estimate_request_context_tokens(api_payload) + if est_tokens > 100_000: + return max(timeout, 600.0) + if est_tokens > 50_000: + return max(timeout, 360.0) + if est_tokens > 25_000: + return max(timeout, 240.0) + if est_tokens > 10_000: + return max(timeout, 180.0) + return timeout + + +def _mark_local_first_chunk_timeout( + error: Exception, + *, + elapsed: float, + threshold: float, + model: Any, + context_tokens: int, +) -> Exception: + """Preserve the local TTFB watchdog reason across SDK transport wrappers.""" + meta = { + "elapsed": int(elapsed), + "threshold": int(threshold), + "model": str(model or "unknown"), + "context_tokens": int(context_tokens or 0), + } + try: + setattr(error, "_hermes_local_first_chunk_timeout", True) + setattr(error, "_hermes_local_first_chunk_meta", meta) + except Exception: + pass + return error + + +def resolve_stream_stale_timeout(agent, api_kwargs: dict) -> float: + """Resolve the no-chunk timeout for streaming chat completions.""" + _cfg_stale = get_provider_stale_timeout(agent.provider, agent.model) + if _cfg_stale is not None: + _stream_stale_timeout_base = _cfg_stale + _uses_implicit_default = False + else: + _env_stale = os.getenv("HERMES_STREAM_STALE_TIMEOUT") + if _env_stale is not None: + _stream_stale_timeout_base = _env_float("HERMES_STREAM_STALE_TIMEOUT", 180.0) + _uses_implicit_default = False + else: + _stream_stale_timeout_base = 180.0 + _uses_implicit_default = True + + _est_tokens = estimate_request_context_tokens(api_kwargs) + if ( + _uses_implicit_default + and agent.base_url + and is_local_endpoint(agent.base_url) + ): + _local_stale = _local_provider_stream_stale_timeout(api_kwargs) + if _local_stale is not None: + logger.debug( + "Local provider detected (%s) — stream stale timeout set to %.0fs", + agent.base_url, + _local_stale, + ) + return _local_stale + logger.debug( + "Local provider detected (%s) — stale stream timeout disabled", + agent.base_url, + ) + return float("inf") + + if _est_tokens > 100_000: + return max(_stream_stale_timeout_base, 300.0) + if _est_tokens > 50_000: + return max(_stream_stale_timeout_base, 240.0) + return _stream_stale_timeout_base + + def interruptible_api_call(agent, api_kwargs: dict): """ Run the API call in a background thread so the main conversation loop @@ -1987,7 +2142,12 @@ def _on_reasoning(text): raise result["error"] return result["response"] - result = {"response": None, "error": None, "partial_tool_names": []} + result = { + "response": None, + "error": None, + "partial_tool_names": [], + "local_first_chunk_timeout": None, + } # Cross-turn stale-stream circuit breaker (#58962) — see the canonical # comment block above ``_stale_streak()``. Raises past the give-up @@ -2037,6 +2197,10 @@ def _close_request_client_once(reason: str) -> None: first_delta_fired = {"done": False} deltas_were_sent = {"yes": False} # Track if any deltas were fired (for fallback) + # Whether the current attempt has received its first stream chunk/event yet. + # The generic local no-first-chunk (TTFB) watchdog in the poll loop reads + # this to tell "connected but no first byte" apart from a healthy stream. + first_chunk_seen = {"yes": False} # Wall-clock timestamp of the last real streaming chunk. The outer # poll loop uses this to detect stale connections that keep receiving # SSE keep-alive pings but no actual data. @@ -2135,6 +2299,7 @@ def _call_chat_completions(): # Reset stale-stream timer so the detector measures from this # attempt's start, not a previous attempt's last chunk. last_chunk_time["t"] = time.time() + first_chunk_seen["yes"] = False agent._touch_activity("waiting for provider response (streaming)") # Initialize per-attempt stream diagnostics so the retry block can # reach for them after the stream dies. Lives on @@ -2214,6 +2379,7 @@ def _call_chat_completions(): reasoning_parts: list = [] usage_obj = None for chunk in stream: + first_chunk_seen["yes"] = True last_chunk_time["t"] = time.time() agent._touch_activity("receiving stream response") @@ -2502,6 +2668,7 @@ def _call_anthropic(): # Reset stale-stream timer for this attempt last_chunk_time["t"] = time.time() + first_chunk_seen["yes"] = False # Per-attempt diagnostic dict for the retry block to consume. _diag = agent._stream_diag_init() request_client_holder["diag"] = _diag @@ -2526,6 +2693,7 @@ def _call_anthropic(): except Exception: pass for event in stream: + first_chunk_seen["yes"] = True # Update stale-stream timer on every event so the # outer poll loop knows data is flowing. Without # this, the detector kills healthy long-running @@ -2889,42 +3057,25 @@ def _call(): finally: _close_request_client_once("stream_request_complete") - # Provider-configured stale timeout takes priority over env default. - _cfg_stale = get_provider_stale_timeout(agent.provider, agent.model) - if _cfg_stale is not None: - _stream_stale_timeout_base = _cfg_stale - else: - _stream_stale_timeout_base = env_float("HERMES_STREAM_STALE_TIMEOUT", 180.0) - # Local providers (Ollama, oMLX, llama-cpp) can take 300+ seconds - # for prefill on large contexts. Disable the stale detector unless - # the user explicitly set HERMES_STREAM_STALE_TIMEOUT. - if _stream_stale_timeout_base == 180.0 and agent.base_url and is_local_endpoint(agent.base_url): - _stream_stale_timeout = float("inf") - logger.debug("Local provider detected (%s) — stale stream timeout disabled", agent.base_url) - else: - # Scale the stale timeout for large contexts: slow models (like Opus) - # can legitimately think for minutes before producing the first token - # when the context is large. Without this, the stale detector kills - # healthy connections during the model's thinking phase, producing - # spurious RemoteProtocolError ("peer closed connection"). - _est_tokens = estimate_request_context_tokens(api_kwargs) - if _est_tokens > 100_000: - _stream_stale_timeout = max(_stream_stale_timeout_base, 300.0) - elif _est_tokens > 50_000: - _stream_stale_timeout = max(_stream_stale_timeout_base, 240.0) - else: - _stream_stale_timeout = _stream_stale_timeout_base - # Reasoning-model floor: known reasoning models (Nemotron 3 Ultra, - # OpenAI o1/o3, Anthropic Opus 4.x thinking, DeepSeek R1, Qwen QwQ, - # xAI Grok reasoning, etc.) routinely exceed the default 180s chat- - # model threshold during their thinking phase. The cloud gateway - # upstream kills the socket first, surfacing as BrokenPipeError. - # Raises the floor only — never overrides explicit user config - # (handled by get_provider_stale_timeout above). - from agent.reasoning_timeouts import get_reasoning_stale_timeout_floor - _reasoning_floor = get_reasoning_stale_timeout_floor(api_kwargs.get("model")) - if _reasoning_floor is not None: - _stream_stale_timeout = max(_stream_stale_timeout, _reasoning_floor) + _stream_stale_timeout = resolve_stream_stale_timeout(agent, api_kwargs) + # Reasoning-model floor: known reasoning models (Nemotron 3 Ultra, + # OpenAI o1/o3, Anthropic Opus 4.x thinking, DeepSeek R1, Qwen QwQ, + # xAI Grok reasoning, etc.) routinely exceed the default 180s chat- + # model threshold during their thinking phase. The cloud gateway + # upstream kills the socket first, surfacing as BrokenPipeError. + # Raises the floor only — never overrides explicit user config + # (handled by get_provider_stale_timeout above). + from agent.reasoning_timeouts import get_reasoning_stale_timeout_floor + _reasoning_floor = get_reasoning_stale_timeout_floor(api_kwargs.get("model")) + if _reasoning_floor is not None: + _stream_stale_timeout = max(_stream_stale_timeout, _reasoning_floor) + _local_first_chunk_timeout = None + _local_first_chunk_model = api_kwargs.get("model") or agent.model + if agent.base_url and is_local_endpoint(agent.base_url): + _local_first_chunk_timeout = _local_provider_first_chunk_timeout( + api_kwargs, + _local_first_chunk_model, + ) t = threading.Thread(target=_call, daemon=True) t.start() @@ -2949,6 +3100,67 @@ def _call(): f"waiting for stream response ({_waiting_secs}s, no chunks yet)" ) + # Local providers have a distinct no-first-chunk failure mode: the HTTP + # request is accepted but the server never emits the first SSE chunk. + # Do not inherit the larger long-context stale timeout for this phase; + # once any chunk arrives, the normal stale detector below takes over. + if _local_first_chunk_timeout is not None and not first_chunk_seen["yes"]: + _first_elapsed = time.time() - last_chunk_time["t"] + if _first_elapsed > _local_first_chunk_timeout: + _est_ctx = estimate_request_context_tokens(api_kwargs) + _local_label = f"local {_local_first_chunk_model or 'model'}" + logger.warning( + "%s stream produced no first chunk for %.0fs " + "(threshold %.0fs). model=%s context=~%s tokens. " + "Killing connection.", + _local_label.capitalize(), + _first_elapsed, + _local_first_chunk_timeout, + api_kwargs.get("model", "unknown"), + f"{_est_ctx:,}", + ) + agent._buffer_status( + f"⚠️ No first stream chunk from {_local_label} for " + f"{int(_first_elapsed)}s " + f"(context: ~{_est_ctx:,} tokens). Switching fallback..." + ) + result["local_first_chunk_timeout"] = { + "elapsed": _first_elapsed, + "threshold": _local_first_chunk_timeout, + "model": _local_first_chunk_model, + "context_tokens": _est_ctx, + } + try: + _close_request_client_once("local_first_chunk_kill") + except Exception: + pass + try: + agent._replace_primary_openai_client( + reason="local_first_chunk_pool_cleanup" + ) + except Exception: + pass + t.join(timeout=_env_float("HERMES_STREAM_ABORT_JOIN_TIMEOUT", 2.0)) + if result["error"] is None and result["response"] is None: + result["error"] = _mark_local_first_chunk_timeout( + TimeoutError( + f"{_local_label.capitalize()} stream produced no first chunk after " + f"{int(_first_elapsed)}s " + f"(threshold: {int(_local_first_chunk_timeout)}s)" + ), + elapsed=_first_elapsed, + threshold=_local_first_chunk_timeout, + model=_local_first_chunk_model, + context_tokens=_est_ctx, + ) + break + if result["error"] is not None or result["response"] is not None: + break + last_chunk_time["t"] = time.time() + agent._touch_activity( + f"{_local_label} first-chunk timeout after {int(_first_elapsed)}s" + ) + # Detect stale streams: connections kept alive by SSE pings # but delivering no real chunks. Kill the client so the # inner retry loop can start a fresh connection. @@ -3021,6 +3233,15 @@ def _call(): if agent._interrupt_requested: raise InterruptedError("Agent interrupted during streaming API call (post-worker)") if result["error"] is not None: + _first_chunk_meta = result.get("local_first_chunk_timeout") + if _first_chunk_meta: + result["error"] = _mark_local_first_chunk_timeout( + result["error"], + elapsed=float(_first_chunk_meta.get("elapsed") or 0.0), + threshold=float(_first_chunk_meta.get("threshold") or 0.0), + model=_first_chunk_meta.get("model"), + context_tokens=int(_first_chunk_meta.get("context_tokens") or 0), + ) if deltas_were_sent["yes"]: # Streaming failed AFTER some tokens were already delivered to # the platform. Re-raising would let the outer retry loop make diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index ce62ffcde1a91..8b6d27ef7eade 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -3901,6 +3901,25 @@ def _perform_api_call(next_api_kwargs): } if retry_count >= max_retries: + if classified.reason == FailoverReason.local_first_chunk_timeout: + if agent._has_pending_fallback(): + meta = classified.error_context or {} + waited = meta.get("elapsed") + waited_text = ( + f" after {int(waited)}s" + if isinstance(waited, (int, float)) and waited > 0 + else "" + ) + agent._buffer_status( + "⚠️ Local model produced no first chunk" + f"{waited_text} — trying fallback..." + ) + if agent._try_activate_fallback(reason=classified.reason): + retry_count = 0 + compression_attempts = 0 + _retry.primary_recovery_attempted = False + continue + _retry.primary_recovery_attempted = True # Before falling back, try rebuilding the primary # client once for transient transport errors (stale # connection pool, TCP reset). Only attempted once diff --git a/agent/error_classifier.py b/agent/error_classifier.py index 4d75502dab427..d4839ce021060 100644 --- a/agent/error_classifier.py +++ b/agent/error_classifier.py @@ -46,6 +46,7 @@ class FailoverReason(enum.Enum): # Retrying reproduces the identical handshake failure, so fail fast # with actionable guidance instead of burning retries. ssl_cert_verification = "ssl_cert_verification" + local_first_chunk_timeout = "local_first_chunk_timeout" # Local stream accepted request but emitted no first chunk — fail over immediately # Context / payload context_overflow = "context_overflow" # Context too large — compress, not failover @@ -609,6 +610,18 @@ def _result(reason: FailoverReason, **overrides) -> ClassifiedError: # ── 1. Provider-specific patterns (highest priority) ──────────── + # The local-provider TTFB watchdog may force-close the SDK stream, which can + # surface as a generic APIConnectionError. Honor the marker from the + # watchdog so the retry loop fails over instead of rebuilding and waiting + # on the same wedged local path again. + if getattr(error, "_hermes_local_first_chunk_timeout", False): + return _result( + FailoverReason.local_first_chunk_timeout, + retryable=False, + should_fallback=True, + error_context=getattr(error, "_hermes_local_first_chunk_meta", None) or {}, + ) + # Provider content-policy / safety-filter block. The provider has made a # deterministic refusal decision about THIS prompt — retrying unchanged # just reproduces the same refusal and burns paid attempts. Must run diff --git a/run_agent.py b/run_agent.py index 2209a1cb249e1..981e254a3eae0 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1244,9 +1244,9 @@ def _resolved_api_call_stale_timeout_base(self) -> tuple[float, bool]: ``_compute_non_stream_stale_timeout``. Returns ``(timeout_seconds, uses_implicit_default)`` so the caller can - preserve legacy behaviors that only apply when the user has *not* - explicitly configured a stale timeout, such as auto-disabling the - detector for local endpoints. + apply behaviors that only make sense when the user has *not* explicitly + configured a stale timeout, such as using local-backend defaults for + loopback endpoints. """ cfg = get_provider_stale_timeout(self.provider, self.model) if cfg is not None: @@ -1271,6 +1271,18 @@ def _resolved_api_call_stale_timeout_base(self) -> tuple[float, bool]: return 90.0, True + def _has_explicit_api_call_stale_timeout(self) -> bool: + """Whether the user explicitly configured a non-stream stale timeout. + + Explicit config = a provider/model ``stale_timeout_seconds`` value or the + ``HERMES_API_CALL_STALE_TIMEOUT`` env var. The reasoning-model floor is + *not* explicit config: it must not suppress the finite local-endpoint + bound applied in :meth:`_compute_non_stream_stale_timeout`. + """ + if get_provider_stale_timeout(self.provider, self.model) is not None: + return True + return os.getenv("HERMES_API_CALL_STALE_TIMEOUT") is not None + def _compute_non_stream_stale_timeout(self, api_payload: Any) -> float: """Compute the effective non-stream stale timeout for this request. @@ -1279,11 +1291,28 @@ def _compute_non_stream_stale_timeout(self, api_payload: Any) -> float: applies the same way to both shapes via :func:`agent.chat_completion_helpers.estimate_request_context_tokens`. """ - stale_base, uses_implicit_default = self._resolved_api_call_stale_timeout_base() base_url = getattr(self, "_base_url", None) or self.base_url or "" - if uses_implicit_default and base_url and is_local_endpoint(base_url): - return float("inf") + # Local backends get a finite, context-scaled non-stream bound whenever + # the user has not *explicitly* configured a stale timeout. Gate this on + # explicit config alone — NOT on ``uses_implicit_default``, which + # _resolved_api_call_stale_timeout_base also clears for the + # reasoning-model floor. A reasoning model served from a local endpoint + # still needs the finite local bound so a stalled non-stream call can + # fall back, instead of inheriting the cloud reasoning floor. (Mirrors + # the streaming path, whose implicit-default flag is derived from + # provider/env config only.) + if ( + base_url + and is_local_endpoint(base_url) + and not self._has_explicit_api_call_stale_timeout() + ): + from agent.chat_completion_helpers import _local_provider_non_stream_stale_timeout + model = self.model + if isinstance(api_payload, dict): + model = api_payload.get("model") or model + return _local_provider_non_stream_stale_timeout(api_payload, model) + stale_base, _uses_implicit_default = self._resolved_api_call_stale_timeout_base() from agent.chat_completion_helpers import estimate_request_context_tokens est_tokens = estimate_request_context_tokens(api_payload) if est_tokens > 100_000: diff --git a/tests/agent/test_error_classifier.py b/tests/agent/test_error_classifier.py index baadfa7e196af..b96e89262586a 100644 --- a/tests/agent/test_error_classifier.py +++ b/tests/agent/test_error_classifier.py @@ -56,6 +56,7 @@ def test_enum_members_exist(self): "upstream_rate_limit", "overloaded", "server_error", "timeout", "ssl_cert_verification", + "local_first_chunk_timeout", "context_overflow", "payload_too_large", "image_too_large", "model_not_found", "format_error", "invalid_encrypted_content", @@ -976,6 +977,23 @@ def test_timeout_error_builtin(self): result = classify_api_error(e) assert result.reason == FailoverReason.timeout + def test_local_first_chunk_timeout_marker_fails_over_without_retry(self): + e = ConnectError("Connection error.") + e._hermes_local_first_chunk_timeout = True + e._hermes_local_first_chunk_meta = { + "elapsed": 75, + "threshold": 75, + "model": "qwen3.6-27b-256k", + "context_tokens": 19956, + } + + result = classify_api_error(e, provider="local", model="qwen3.6-27b-256k") + + assert result.reason == FailoverReason.local_first_chunk_timeout + assert result.retryable is False + assert result.should_fallback is True + assert result.error_context["elapsed"] == 75 + def test_runtime_error_cli_turn_timed_out_classifies_as_timeout(self): # RuntimeError from a local claude-cli shim that wraps a subprocess # timeout must classify as FailoverReason.timeout, not unknown, so diff --git a/tests/agent/test_local_stream_timeout.py b/tests/agent/test_local_stream_timeout.py index 91ca7f404c61a..9a51e477aab50 100644 --- a/tests/agent/test_local_stream_timeout.py +++ b/tests/agent/test_local_stream_timeout.py @@ -11,6 +11,13 @@ from unittest.mock import patch from agent.model_metadata import is_local_endpoint +from agent.chat_completion_helpers import ( + _local_provider_first_chunk_timeout, + _local_provider_non_stream_stale_timeout, + _local_provider_stream_stale_timeout, + _mark_local_first_chunk_timeout, + resolve_stream_stale_timeout, +) class TestLocalStreamReadTimeout: @@ -73,6 +80,183 @@ def test_empty_base_url_keeps_default(self): assert _stream_read_timeout == 120.0 +class TestLocalStaleTimeout: + """Local backends keep unbounded stale behavior by default, with an opt-in bound.""" + + @staticmethod + def _payload_for_estimated_tokens(tokens: int) -> dict[str, list[str]]: + return {"messages": ["x" * (tokens * 4)]} + + def _make_agent(self, *, model="qwen3.6-27b-256k", base_url="http://127.0.0.1:8080/v1"): + from run_agent import AIAgent + + with patch("agent.context_compressor.get_model_context_length", return_value=131072): + return AIAgent( + api_key="sk-dummy", + base_url=base_url, + provider="taro", + model=model, + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + platform="cli", + ) + + def test_opt_in_local_stream_stale_timeout_bounds_watchdog(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / ".env").write_text("", encoding="utf-8") + monkeypatch.delenv("HERMES_STREAM_STALE_TIMEOUT", raising=False) + monkeypatch.setenv("HERMES_LOCAL_STALE_TIMEOUT", "75") + + agent = self._make_agent() + + timeout = resolve_stream_stale_timeout( + agent, + {"model": "qwen3.6-27b-256k", "messages": [{"role": "user", "content": "hi"}]}, + ) + + assert timeout == 75.0 + + def test_local_stream_stale_timeout_is_opt_in(self, monkeypatch): + monkeypatch.delenv("HERMES_LOCAL_STALE_TIMEOUT", raising=False) + monkeypatch.delenv("HERMES_LOCAL_STREAM_STALE_TIMEOUT", raising=False) + + assert _local_provider_stream_stale_timeout({"messages": []}) is None + + def test_generic_local_first_chunk_timeout_is_finite(self, monkeypatch): + monkeypatch.delenv("HERMES_LOCAL_FIRST_CHUNK_TIMEOUT", raising=False) + monkeypatch.delenv("HERMES_LOCAL_TTFB_TIMEOUT", raising=False) + + timeout = _local_provider_first_chunk_timeout( + self._payload_for_estimated_tokens(6_000), + "qwen3.6-27b-256k", + ) + + assert timeout == 90.0 + + def test_generic_local_first_chunk_timeout_scales_for_large_context(self, monkeypatch): + monkeypatch.setenv("HERMES_LOCAL_FIRST_CHUNK_TIMEOUT", "120") + + timeout = _local_provider_first_chunk_timeout( + self._payload_for_estimated_tokens(66_000), + "qwen3.6-27b-256k", + ) + + assert timeout == 360.0 + + def test_local_first_chunk_timeout_marker_preserves_watchdog_metadata(self): + err = RuntimeError("Connection error.") + + marked = _mark_local_first_chunk_timeout( + err, + elapsed=180.4, + threshold=180.0, + model="qwen3.6-27b-256k", + context_tokens=42000, + ) + + assert marked is err + assert getattr(err, "_hermes_local_first_chunk_timeout") is True + assert getattr(err, "_hermes_local_first_chunk_meta") == { + "elapsed": 180, + "threshold": 180, + "model": "qwen3.6-27b-256k", + "context_tokens": 42000, + } + + def test_generic_local_stream_stale_timeout_still_disables_by_default(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / ".env").write_text("", encoding="utf-8") + monkeypatch.delenv("HERMES_STREAM_STALE_TIMEOUT", raising=False) + + agent = self._make_agent(model="qwen3.6-27b") + + timeout = resolve_stream_stale_timeout( + agent, + {"model": "qwen3.6-27b", "messages": [{"role": "user", "content": "hi"}]}, + ) + + assert timeout == float("inf") + + def test_local_non_stream_stale_timeout_env_override(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / ".env").write_text("", encoding="utf-8") + monkeypatch.delenv("HERMES_API_CALL_STALE_TIMEOUT", raising=False) + monkeypatch.setenv("HERMES_LOCAL_NON_STREAM_STALE_TIMEOUT", "75") + + agent = self._make_agent() + + assert agent._compute_non_stream_stale_timeout({"messages": []}) == 75.0 + + def test_generic_local_non_stream_stale_timeout_is_finite(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / ".env").write_text("", encoding="utf-8") + monkeypatch.delenv("HERMES_API_CALL_STALE_TIMEOUT", raising=False) + monkeypatch.delenv("HERMES_LOCAL_NON_STREAM_STALE_TIMEOUT", raising=False) + monkeypatch.delenv("HERMES_LOCAL_RESPONSE_TIMEOUT", raising=False) + + agent = self._make_agent(model="qwen3.6-27b-256k") + + assert agent._compute_non_stream_stale_timeout({"messages": []}) == 120.0 + + def test_generic_local_non_stream_stale_timeout_scales(self, monkeypatch): + monkeypatch.setenv("HERMES_LOCAL_NON_STREAM_STALE_TIMEOUT", "120") + + timeout = _local_provider_non_stream_stale_timeout( + self._payload_for_estimated_tokens(66_000), + "qwen3.6-27b-256k", + ) + + assert timeout == 360.0 + + def test_explicit_stream_stale_timeout_wins_over_local_opt_in(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / ".env").write_text("", encoding="utf-8") + monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "12") + monkeypatch.setenv("HERMES_LOCAL_STALE_TIMEOUT", "75") + + agent = self._make_agent() + + assert resolve_stream_stale_timeout(agent, {"model": "qwen3.6-27b-256k", "messages": []}) == 12.0 + + @pytest.mark.parametrize( + ("estimated_tokens", "expected_timeout"), + [ + (10_000, 75.0), + (10_001, 90.0), + (25_000, 90.0), + (25_001, 150.0), + (50_000, 150.0), + (50_001, 240.0), + (100_000, 240.0), + (100_001, 300.0), + ], + ) + def test_opt_in_local_stale_timeout_threshold_boundaries( + self, + monkeypatch, + estimated_tokens, + expected_timeout, + ): + monkeypatch.setenv("HERMES_LOCAL_STALE_TIMEOUT", "75") + monkeypatch.delenv("HERMES_LOCAL_STREAM_STALE_TIMEOUT", raising=False) + + timeout = _local_provider_stream_stale_timeout( + self._payload_for_estimated_tokens(estimated_tokens), + ) + + assert timeout == expected_timeout + + def test_non_positive_local_stale_timeout_disables_watchdog(self, monkeypatch): + monkeypatch.setenv("HERMES_LOCAL_STALE_TIMEOUT", "0") + + timeout = _local_provider_stream_stale_timeout( + self._payload_for_estimated_tokens(1), + ) + + assert timeout == float("inf") + + class TestIsLocalEndpoint: """Direct unit tests for is_local_endpoint.""" diff --git a/tests/hermes_cli/test_timeouts.py b/tests/hermes_cli/test_timeouts.py index 93c8cafc0a90b..51d5fb6d49246 100644 --- a/tests/hermes_cli/test_timeouts.py +++ b/tests/hermes_cli/test_timeouts.py @@ -268,10 +268,12 @@ def test_resolved_api_call_stale_timeout_priority(monkeypatch, tmp_path): assert agent2._resolved_api_call_stale_timeout_base() == (90.0, True) -def test_default_non_stream_stale_timeout_auto_disables_for_local_endpoints(monkeypatch, tmp_path): +def test_default_non_stream_stale_timeout_is_finite_for_local_endpoints(monkeypatch, tmp_path): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) (tmp_path / ".env").write_text("", encoding="utf-8") monkeypatch.delenv("HERMES_API_CALL_STALE_TIMEOUT", raising=False) + monkeypatch.delenv("HERMES_LOCAL_NON_STREAM_STALE_TIMEOUT", raising=False) + monkeypatch.delenv("HERMES_LOCAL_RESPONSE_TIMEOUT", raising=False) from run_agent import AIAgent agent = AIAgent( @@ -285,7 +287,7 @@ def test_default_non_stream_stale_timeout_auto_disables_for_local_endpoints(monk platform="cli", ) - assert agent._compute_non_stream_stale_timeout([]) == float("inf") + assert agent._compute_non_stream_stale_timeout([]) == 120.0 def test_explicit_non_stream_stale_timeout_is_honored_for_local_endpoints(monkeypatch, tmp_path): diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index dfe77d007c1e0..03fc5fc32810f 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -7562,6 +7562,18 @@ def test_no_unreachable_max_retries_after_backoff(self): f"but found {occurrences}" ) + def test_local_first_chunk_timeout_fallback_precedes_primary_recovery(self): + import inspect + from agent.conversation_loop import run_conversation as _rc + source = inspect.getsource(_rc) + + first_chunk_fallback = source.index( + "FailoverReason.local_first_chunk_timeout" + ) + primary_recovery = source.index("agent._try_recover_primary_transport") + + assert first_chunk_fallback < primary_recovery + class TestSupportsReasoningExtraBody: def _make_agent(self):