diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index a0a08649762e..54c7e4678830 100644 --- a/agent/auxiliary_client.py +++ b/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..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[]()`` — 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, diff --git a/agent/context_compressor.py b/agent/context_compressor.py index b652d4ceb2d4..d53c5d5a5488 100644 --- a/agent/context_compressor.py +++ b/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() + "..." diff --git a/tests/agent/test_compression_fallback_budget.py b/tests/agent/test_compression_fallback_budget.py new file mode 100644 index 000000000000..1bae37ad8cfd --- /dev/null +++ b/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..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 diff --git a/website/docs/user-guide/features/fallback-providers.md b/website/docs/user-guide/features/fallback-providers.md index 4adbc72901cb..5355d7a42811 100644 --- a/website/docs/user-guide/features/fallback-providers.md +++ b/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):