feat(agent): finite first-chunk timeout + failover for local OpenAI-compatible streams

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:

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

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

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:

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

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."""
 

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):