fix(compression): give fallback candidates their own timeout budget + escalate repeat-timeout cooldowns (#65143)

Fixes #62452. Two amplifiers turned one slow auxiliary route into a
per-turn multi-minute stall:

1. Fallback candidates inherited the exact effective_timeout the primary
   was called with. When the primary's deadline was short (tuned or
   already burned), an independently healthy fallback died on the same
   clock — the reporter's 163k-token compression needed ~90s on the
   fallback and got the primary's 30s, every turn. fallback_chain
   entries may now declare their own 'timeout' (seconds); both fallback
   candidate call sites (sync + async) resolve it via
   _fallback_entry_timeout, label-scoped so only configured-chain
   candidates are affected. No entry timeout → task-level timeout,
   preserving existing behavior.

2. A session whose transcript structurally cannot be summarized within
   the deadline re-attempted every 60s, re-burning the full timeout on
   every subsequent turn. Consecutive timeout-class failures now
   escalate the cooldown 60s → 300s → 900s (capped); any successful
   summary or session reset clears the streak. Timeout classification
   takes precedence over the streaming-closed 30s rung ('timed out'
   also matches _is_connection_error) and now recognizes the SDK's
   'Request timed out.' phrasing.

Fail-safe behavior is unchanged: all messages are preserved when every
candidate fails; the cooldown only spaces out retries.

agent/auxiliary_client.py

@@ -44,6 +44,7 @@ import contextlib
 import json
 import logging
 import os
+import re
 import threading
 import time
 from pathlib import Path  # noqa: F401 — used by test mocks
@@ -3693,6 +3694,40 @@ def _auth_refresh_provider_for_route(
     return normalized
 
 
+def _fallback_entry_timeout(task: Optional[str], fb_label: str) -> Optional[float]:
+    """Resolve a per-entry ``timeout`` for a configured fallback candidate.
+
+    A fallback candidate previously inherited the exact timeout the primary
+    provider was called with. When that deadline was tuned for the primary
+    (or the primary simply consumed its whole budget before failing over),
+    the fallback aborted on the same clock even when independently healthy —
+    a 163k-token compression that needs ~90s on the fallback died at the
+    primary's 30s deadline every turn (#62452).
+
+    Entries in ``auxiliary.<task>.fallback_chain`` may declare their own
+    ``timeout`` (seconds). This helper reads it by parsing the entry index
+    out of the label minted by :func:`_try_configured_fallback_chain`
+    (``fallback_chain[<i>](<provider>)`` — our own stable format). Returns
+    ``None`` when the label is not a configured-chain candidate, the entry
+    has no ``timeout``, or the value is invalid — callers then keep the
+    task-level timeout, preserving existing behavior.
+    """
+    if not task or not fb_label:
+        return None
+    m = re.match(r"fallback_chain\[(\d+)\]", fb_label)
+    if not m:
+        return None
+    try:
+        chain = _get_auxiliary_task_config(task).get("fallback_chain")
+        entry = chain[int(m.group(1))] if isinstance(chain, list) else None
+        raw = entry.get("timeout") if isinstance(entry, dict) else None
+    except Exception:
+        return None
+    if isinstance(raw, (int, float)) and not isinstance(raw, bool) and raw > 0:
+        return float(raw)
+    return None
+
+
 def _call_fallback_candidate_sync(
     fb_client: Any,
     fb_model: Optional[str],
@@ -3721,7 +3756,20 @@ def _call_fallback_candidate_sync(
     once with a rebuilt client; if the retry also auth-fails (non-refreshable
     expired token), mark the provider unhealthy and return ``None`` so the
     caller can continue to the next fallback layer. Non-auth errors raise.
+
+    ``effective_timeout`` is the task-level deadline; a configured-chain
+    candidate with its own ``timeout`` entry gets that instead, so a
+    fallback tuned differently from the primary is allowed its own budget
+    (#62452).
     """
+    fb_timeout = _fallback_entry_timeout(task, fb_label)
+    if fb_timeout is not None and fb_timeout != effective_timeout:
+        logger.info(
+            "Auxiliary %s: %s using its configured timeout %.0fs "
+            "(task-level was %.0fs)",
+            task or "call", fb_label, fb_timeout, effective_timeout,
+        )
+        effective_timeout = fb_timeout
     fb_base = str(getattr(fb_client, "base_url", "") or "")
     fb_kwargs = _build_call_kwargs(
         fb_label, fb_model, messages,
@@ -3780,6 +3828,14 @@ async def _call_fallback_candidate_async(
     reasoning_config: Optional[dict],
 ) -> Optional[Any]:
     """Async mirror of :func:`_call_fallback_candidate_sync`."""
+    fb_timeout = _fallback_entry_timeout(task, fb_label)
+    if fb_timeout is not None and fb_timeout != effective_timeout:
+        logger.info(
+            "Auxiliary %s: %s using its configured timeout %.0fs "
+            "(task-level was %.0fs)",
+            task or "call", fb_label, fb_timeout, effective_timeout,
+        )
+        effective_timeout = fb_timeout
     fb_base = str(getattr(fb_client, "base_url", "") or "")
     fb_kwargs = _build_call_kwargs(
         fb_label, fb_model, messages,

agent/context_compressor.py

@@ -795,6 +795,7 @@ class ContextCompressor(ContextEngine):
         self._context_probe_persistable = False
         self._previous_summary = None
         self._last_summary_error = None
+        self._consecutive_timeout_failures = 0
         self._last_summary_dropped_count = 0
         self._last_summary_fallback_used = False
         self._last_aux_model_failure_error = None
@@ -833,6 +834,7 @@ class ContextCompressor(ContextEngine):
         """
         self._previous_summary = None
         self._last_summary_error = None
+        self._consecutive_timeout_failures = 0
         self._last_summary_dropped_count = 0
         self._last_summary_fallback_used = False
         self._last_aux_model_failure_error = None
@@ -857,6 +859,7 @@ class ContextCompressor(ContextEngine):
         self._session_id = session_id or ""
         self._summary_failure_cooldown_until = 0.0
         self._last_summary_error = None
+        self._consecutive_timeout_failures = 0
         self._fallback_compression_streak = 0
         self.get_active_compression_failure_cooldown()
         self._load_fallback_compression_streak()
@@ -1005,6 +1008,7 @@ class ContextCompressor(ContextEngine):
     def _clear_compression_failure_cooldown(self) -> None:
         self._summary_failure_cooldown_until = 0.0
         self._last_summary_error = None
+        self._consecutive_timeout_failures = 0
 
         session_db = getattr(self, "_session_db", None)
         session_id = getattr(self, "_session_id", "")
@@ -2288,6 +2292,7 @@ This compaction should PRIORITISE preserving all information related to the focu
             _is_timeout = (
                 _status in {408, 429, 502, 504}
                 or "timeout" in _err_str
+                or "timed out" in _err_str
             )
             # Non-JSON / malformed-body responses from misconfigured providers
             # or proxies (e.g. an HTML 502 page returned with
@@ -2377,7 +2382,30 @@ This compaction should PRIORITISE preserving all information related to the focu
             # Transient errors (timeout, rate limit, network, JSON decode,
             # streaming premature-close) — shorter cooldown for JSON decode and
             # streaming-closed since those conditions can self-resolve quickly.
-            _transient_cooldown = 30 if (_is_json_decode or _is_streaming_closed) else 60
+            # Timeout-class failures escalate with consecutive occurrences:
+            # a session whose transcript structurally exceeds what the
+            # summary route can produce within its deadline will fail the
+            # same way every time, and re-burning the full timeout every
+            # 60s turns each subsequent turn into a multi-minute stall
+            # (#62452). 60s → 300s → 900s (capped); any successful summary
+            # resets the streak via _clear_compression_failure_cooldown().
+            # Timeout takes precedence over the streaming-closed short rung:
+            # a "timed out" error also matches _is_connection_error, but a
+            # deadline exhaustion is the structural repeat-offender class,
+            # not a transient mid-stream drop.
+            if _is_timeout:
+                self._consecutive_timeout_failures = (
+                    getattr(self, "_consecutive_timeout_failures", 0) + 1
+                )
+                _TIMEOUT_COOLDOWN_LADDER = (60, 300, 900)
+                _transient_cooldown = _TIMEOUT_COOLDOWN_LADDER[
+                    min(self._consecutive_timeout_failures,
+                        len(_TIMEOUT_COOLDOWN_LADDER)) - 1
+                ]
+            elif _is_json_decode or _is_streaming_closed:
+                _transient_cooldown = 30
+            else:
+                _transient_cooldown = 60
             err_text = str(e).strip() or e.__class__.__name__
             if len(err_text) > 220:
                 err_text = err_text[:217].rstrip() + "..."

tests/agent/test_compression_fallback_budget.py

@@ -0,0 +1,202 @@
+"""Tests for #62452 — compression fallback timeout independence + escalating
+timeout cooldown.
+
+Two failure amplifiers when the auxiliary compression route times out:
+
+1. Fallback candidates inherited the primary's exact ``effective_timeout``.
+   A fallback tuned differently (or simply slower-but-healthy) died on the
+   same clock the primary just burned, guaranteeing chain exhaustion.
+   Fix: ``auxiliary.<task>.fallback_chain`` entries may declare their own
+   ``timeout``; ``_call_fallback_candidate_sync/async`` resolve it via
+   ``_fallback_entry_timeout``.
+
+2. A session whose transcript structurally cannot be summarised within the
+   deadline re-attempted every 60s, re-burning the full timeout on every
+   subsequent turn. Fix: consecutive timeout-class failures escalate the
+   cooldown 60s → 300s → 900s; any successful summary resets the streak.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from agent.auxiliary_client import _fallback_entry_timeout, _call_fallback_candidate_sync
+from agent.context_compressor import ContextCompressor
+
+
+# ---------------------------------------------------------------------------
+# _fallback_entry_timeout — label parsing + config resolution
+# ---------------------------------------------------------------------------
+
+
+def _patch_task_config(chain):
+    return patch(
+        "agent.auxiliary_client._get_auxiliary_task_config",
+        return_value={"fallback_chain": chain},
+    )
+
+
+def test_entry_timeout_resolved_from_configured_chain():
+    chain = [
+        {"provider": "custom", "timeout": 240},
+        {"provider": "openrouter"},
+    ]
+    with _patch_task_config(chain):
+        assert _fallback_entry_timeout("compression", "fallback_chain[0](custom)") == 240.0
+        # Entry without a timeout → None (keep task-level).
+        assert _fallback_entry_timeout("compression", "fallback_chain[1](openrouter)") is None
+
+
+def test_entry_timeout_ignores_non_chain_labels_and_bad_values():
+    chain = [{"provider": "custom", "timeout": "fast"}]  # invalid type
+    with _patch_task_config(chain):
+        # Non-chain labels (main-model fallback, payment fallback, ...) pass through.
+        assert _fallback_entry_timeout("compression", "anthropic") is None
+        assert _fallback_entry_timeout("compression", "") is None
+        assert _fallback_entry_timeout(None, "fallback_chain[0](custom)") is None
+        # Invalid timeout value → None.
+        assert _fallback_entry_timeout("compression", "fallback_chain[0](custom)") is None
+    # Out-of-range index → None, never raises.
+    with _patch_task_config([]):
+        assert _fallback_entry_timeout("compression", "fallback_chain[5](x)") is None
+    # Boolean True is not a valid timeout (bool is an int subclass).
+    with _patch_task_config([{"provider": "x", "timeout": True}]):
+        assert _fallback_entry_timeout("compression", "fallback_chain[0](x)") is None
+
+
+def test_fallback_candidate_call_uses_entry_timeout():
+    """The wire call for a configured-chain candidate carries the entry's own
+    timeout, not the task-level one the primary just burned."""
+    seen = {}
+
+    class _FakeCompletions:
+        def create(self, **kwargs):
+            seen.update(kwargs)
+            return SimpleNamespace(
+                choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))]
+            )
+
+    fb_client = SimpleNamespace(
+        base_url="https://example.invalid/v1",
+        chat=SimpleNamespace(completions=_FakeCompletions()),
+    )
+    chain = [{"provider": "custom", "timeout": 240}]
+    with _patch_task_config(chain):
+        resp = _call_fallback_candidate_sync(
+            fb_client, "deepseek-v4-flash", "fallback_chain[0](custom)",
+            task="compression", messages=[{"role": "user", "content": "hi"}],
+            temperature=None, max_tokens=None, tools=None,
+            effective_timeout=30.0,  # the primary's burned budget
+            effective_extra_body={}, reasoning_config=None,
+        )
+    assert resp is not None
+    assert seen.get("timeout") == 240.0
+
+
+def test_fallback_candidate_without_entry_timeout_keeps_task_timeout():
+    seen = {}
+
+    class _FakeCompletions:
+        def create(self, **kwargs):
+            seen.update(kwargs)
+            return SimpleNamespace(
+                choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))]
+            )
+
+    fb_client = SimpleNamespace(
+        base_url="https://example.invalid/v1",
+        chat=SimpleNamespace(completions=_FakeCompletions()),
+    )
+    with _patch_task_config([{"provider": "custom"}]):
+        _call_fallback_candidate_sync(
+            fb_client, "m", "fallback_chain[0](custom)",
+            task="compression", messages=[{"role": "user", "content": "hi"}],
+            temperature=None, max_tokens=None, tools=None,
+            effective_timeout=300.0,
+            effective_extra_body={}, reasoning_config=None,
+        )
+    assert seen.get("timeout") == 300.0
+
+
+# ---------------------------------------------------------------------------
+# Escalating timeout cooldown — 60s → 300s → 900s, reset on success
+# ---------------------------------------------------------------------------
+
+
+def _make_compressor():
+    with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+        return ContextCompressor(model="main-model", quiet_mode=True)
+
+
+def _msgs():
+    return [
+        {"role": "user", "content": "u1 " + "x" * 200},
+        {"role": "assistant", "content": "a1 " + "y" * 200},
+        {"role": "user", "content": "u2 " + "z" * 200},
+    ]
+
+
+def _fail_with_timeout(compressor, now):
+    with patch(
+        "agent.context_compressor.call_llm",
+        side_effect=TimeoutError("Request timed out."),
+    ), patch("agent.context_compressor.time.monotonic", return_value=now):
+        return compressor._generate_summary(_msgs())
+
+
+def test_timeout_cooldown_escalates_and_caps():
+    c = _make_compressor()
+
+    assert _fail_with_timeout(c, 1000.0) is None
+    assert c._summary_failure_cooldown_until == 1000.0 + 60
+
+    assert _fail_with_timeout(c, 2000.0) is None
+    assert c._summary_failure_cooldown_until == 2000.0 + 300
+
+    assert _fail_with_timeout(c, 3000.0) is None
+    assert c._summary_failure_cooldown_until == 3000.0 + 900
+
+    # Capped: a fourth consecutive timeout stays at the ladder max.
+    assert _fail_with_timeout(c, 4000.0) is None
+    assert c._summary_failure_cooldown_until == 4000.0 + 900
+
+
+def test_timeout_streak_resets_on_success():
+    c = _make_compressor()
+    assert _fail_with_timeout(c, 1000.0) is None
+    assert _fail_with_timeout(c, 2000.0) is None
+    assert c._consecutive_timeout_failures == 2
+
+    # A successful summary clears the cooldown AND the streak.
+    c._clear_compression_failure_cooldown()
+    assert c._consecutive_timeout_failures == 0
+
+    # The next timeout starts back at the 60s rung.
+    assert _fail_with_timeout(c, 5000.0) is None
+    assert c._summary_failure_cooldown_until == 5000.0 + 60
+
+
+def test_non_timeout_transient_errors_keep_flat_cooldown():
+    """Rate-limit / generic connection errors keep the flat 60s cooldown —
+    escalation is scoped to timeout-class failures only."""
+    c = _make_compressor()
+    with patch(
+        "agent.context_compressor.call_llm",
+        side_effect=RuntimeError("rate limit exceeded"),
+    ), patch("agent.context_compressor.time.monotonic", return_value=1000.0):
+        assert c._generate_summary(_msgs()) is None
+    assert c._summary_failure_cooldown_until == 1000.0 + 60
+
+    # And it does not advance the timeout streak.
+    assert getattr(c, "_consecutive_timeout_failures", 0) == 0
+
+
+def test_session_reset_clears_timeout_streak():
+    c = _make_compressor()
+    assert _fail_with_timeout(c, 1000.0) is None
+    assert _fail_with_timeout(c, 2000.0) is None
+    assert c._consecutive_timeout_failures == 2
+
+    c.on_session_reset()
+    assert c._consecutive_timeout_failures == 0

website/docs/user-guide/features/fallback-providers.md

@@ -341,10 +341,13 @@ auxiliary:
     fallback_chain:
       - provider: openai
         model: gpt-4o-mini
+        timeout: 240            # optional — this candidate's own deadline (seconds)
 ```
 
 You do **not** need to configure `fallback_chain` to get fallback — the main-agent safety net runs regardless. Use it only when you specifically want a different order than the default.
 
+Each `fallback_chain` entry may also declare its own `timeout` (seconds). Without it, a fallback candidate inherits the task-level timeout — which may be tuned for the primary provider. Declaring a per-entry `timeout` lets a slower-but-reliable fallback (e.g. a large-context summarizer) get the budget it actually needs instead of dying on the primary's clock.
+
 ### Provider quota errors that trigger fallback
 
 Hermes recognizes these as capacity-equivalent to 402 credit exhaustion (not transient rate limits):