From 951600dc79d658f283cea1000a69be8df2cdb547 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Thu, 17 Sep 2026 01:35:48 -0700 Subject: [PATCH] fix(aux): quarantine a transient fallback-candidate failure for seconds, not the payment hold A per-minute 429, a timeout or a dropped connection on one fallback_providers lane went through _quarantine_fallback_candidate -> _mark_provider_unhealthy with the default 600 s TTL, hiding that lane process-wide from every aux task for ten minutes and logging it as a "payment / credit error". The class of the failure now picks the hold: agent/auxiliary_health.py::fallback_candidate_quarantine_ttl is a table keyed by the _FALLBACK_REASONS label (rate limit / connection / invalid response -> 60 s; payment/quota and a dead credential keep the long default), and _mark_provider_unhealthy takes the reason so the warning names the real class. Sync and async candidate helpers pass the label instead of a pre-rendered `why` string. Review follow-ups on #113965: the exhaustion log, module docstring and T4 test name now say the PRIMARY (narrowed) error is raised, which is what happens; the multihop tests no longer mock _mark_provider_unhealthy (they reset the unhealthy cache and assert the real per-endpoint quarantine), and the stale-candidate test asserts the second chain walk happened instead of pinning the re-selection reason label. One invariant test added: a plain 429 on a candidate is held <= 60 s with no "payment" wording, a 402 candidate keeps the long hold (red on the previous head by source-file swap: 599.99 s). --- agent/auxiliary_client.py | 21 +++--- agent/auxiliary_health.py | 18 +++++ tests/agent/test_auxiliary_client.py | 9 +-- .../agent/test_auxiliary_fallback_multihop.py | 69 ++++++++++++++++--- 4 files changed, 95 insertions(+), 22 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 2806872968..8caace4019 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -111,7 +111,8 @@ from agent.credential_pool import load_pool from agent.model_metadata import MINIMUM_CONTEXT_LENGTH, get_model_context_length from hermes_cli.config import get_hermes_home from agent.auxiliary_health import ( - _custom_health_base_url, _unhealthy_cache_key, fallback_candidate_unavailable_reason, + _custom_health_base_url, _unhealthy_cache_key, fallback_candidate_quarantine_ttl, + fallback_candidate_unavailable_reason, ) from agent.auxiliary_unavailable import ( AuxiliaryClientUnavailable, clear_nous_credential_failure, nous_credential_failure_detail, @@ -3923,11 +3924,15 @@ def _plan_fallback_candidate( def _quarantine_fallback_candidate( task: Optional[str], fb_label: str, fb_provider: str, fb_err: Exception, *, - base_url: str = "", tag: str = "", why: str = "has a stale/unrefreshable credential", + base_url: str = "", tag: str = "", reason: Optional[str] = None, ) -> None: - """The candidate cannot serve this walk (dead token, or a capacity error such as a quota 429): - mark it unhealthy so the ordered re-walk skips it and the caller moves on to the next entry.""" - _mark_provider_unhealthy(fb_provider or fb_label, base_url=base_url, reason="stale fallback credential") + """The candidate cannot serve this walk (``reason`` = its ``_FALLBACK_REASONS`` capacity label, + None = dead token): mark it unhealthy so the ordered re-walk skips it and the caller moves on to + the next entry. Transient classes get a short hold, payment/quota and dead tokens the long one.""" + _mark_provider_unhealthy( + fb_provider or fb_label, ttl=fallback_candidate_quarantine_ttl(reason), + base_url=base_url, reason=reason or "stale fallback credential") + why = f"is out of capacity ({reason})" if reason else "has a stale/unrefreshable credential" logger.warning("Auxiliary %s%s: fallback candidate %s %s (%s) — skipping to next fallback", task or "call", tag, fb_label, why, fb_err) @@ -4000,7 +4005,7 @@ def _call_fallback_candidate_sync( raise _quarantine_fallback_candidate( task, fb_label, destination.provider, fb_err, base_url=destination.base_url, - why=f"is out of capacity ({capacity})") + reason=capacity) return None fb_provider, retry = _plan_fallback_auth_retry( destination, rebuild, async_mode=False, failed_api_key=getattr(fb_client, "api_key", "")) @@ -4050,7 +4055,7 @@ async def _call_fallback_candidate_async( raise _quarantine_fallback_candidate( task, fb_label, destination.provider, fb_err, base_url=destination.base_url, - tag=" (async)", why=f"is out of capacity ({capacity})") + tag=" (async)", reason=capacity) return None fb_provider, retry = _plan_fallback_auth_retry( destination, rebuild, async_mode=True, failed_api_key=getattr(fb_client, "api_key", "")) @@ -7493,7 +7498,7 @@ def _ladder_provider_fallback(first_err: Exception, route: _LadderRoute): # All fallback layers exhausted — emit a single user-visible warning so the operator # knows aux task is about to fail. (#26882) The error itself is re-raised below. # (#26882) - "(fallback_chain + main agent model). Raising the last error.", + "(fallback_chain + main agent model). Raising the primary error.", task or "call", tag, reason, resolved_provider) return None diff --git a/agent/auxiliary_health.py b/agent/auxiliary_health.py index 7301f33782..0653a59241 100644 --- a/agent/auxiliary_health.py +++ b/agent/auxiliary_health.py @@ -53,3 +53,21 @@ def fallback_candidate_unavailable_reason(exc: Exception) -> Optional[str]: (label for predicate, label in _FALLBACK_REASONS if label != "auth error" and predicate(exc)), None, ) + + +# Quarantine hold per unavailable-reason label. Payment/quota depletion and a dead credential +# last hours, so those keep the long default TTL (None); a per-minute 429, a dropped connection or +# a garbled body clears in seconds — holding the lane for 10 minutes process-wide would hide a +# healthy fallback from every aux task over one transient blip. +_TRANSIENT_CANDIDATE_QUARANTINE_SECONDS = 60.0 +_CANDIDATE_QUARANTINE_TTL: dict[str, Optional[float]] = { + "rate limit": _TRANSIENT_CANDIDATE_QUARANTINE_SECONDS, + "connection error": _TRANSIENT_CANDIDATE_QUARANTINE_SECONDS, + "invalid provider response": _TRANSIENT_CANDIDATE_QUARANTINE_SECONDS, +} + + +def fallback_candidate_quarantine_ttl(reason: Optional[str]) -> Optional[float]: + """Seconds to hide a fallback candidate for ``reason`` (a ``_FALLBACK_REASONS`` label, or None + for a stale credential); None means the long default TTL.""" + return _CANDIDATE_QUARANTINE_TTL.get(reason or "") diff --git a/tests/agent/test_auxiliary_client.py b/tests/agent/test_auxiliary_client.py index 1b8e8784f6..c9baf11550 100644 --- a/tests/agent/test_auxiliary_client.py +++ b/tests/agent/test_auxiliary_client.py @@ -1834,11 +1834,12 @@ class TestStaleFallbackCandidateSkip: ) assert result.choices[0].message.content == "openrouter-serves" + # The chain was walked a second time after the stale candidate was quarantined. assert mock_fb.call_count == 2 - assert mock_fb.call_args_list[1].kwargs.get("reason") == "fallback candidate unavailable" - mock_mark.assert_called_once_with( - "anthropic", base_url="https://api.anthropic.com", reason="stale fallback credential", - ) + assert mock_mark.call_count == 1 + assert mock_mark.call_args.args[0] == "anthropic" + assert mock_mark.call_args.kwargs["base_url"] == "https://api.anthropic.com" + assert mock_mark.call_args.kwargs["reason"] == "stale fallback credential" assert stale_fb.chat.completions.create.call_count == 1 assert healthy_fb.chat.completions.create.call_count == 1 diff --git a/tests/agent/test_auxiliary_fallback_multihop.py b/tests/agent/test_auxiliary_fallback_multihop.py index a0878dec15..6f723a785e 100644 --- a/tests/agent/test_auxiliary_fallback_multihop.py +++ b/tests/agent/test_auxiliary_fallback_multihop.py @@ -3,15 +3,26 @@ When a ``fallback_providers`` candidate itself fails with a quota/rate-limit/payment/capacity error, the walk must advance to the next configured entry instead of letting the candidate's exception escape after a single hop. Every lane is attempted at most once; when the whole chain -is exhausted the last error still surfaces as a controlled failure. Sync and async share the walk. +is exhausted the primary (narrowed) error still surfaces as a controlled failure. Sync and async +share the walk. A quarantined candidate is hidden for a TTL that matches its failure class: seconds +for a transient 429 / dropped connection, the long payment hold only for depleted credit. """ +import time from unittest.mock import AsyncMock, MagicMock, patch import pytest +from agent import auxiliary_client as ac from agent.auxiliary_client import async_call_llm, call_llm +@pytest.fixture(autouse=True) +def _fresh_unhealthy_cache(): + ac._reset_aux_unhealthy_cache() + yield + ac._reset_aux_unhealthy_cache() + + def _quota_429(lane: str) -> Exception: exc = Exception( f"Error code: 429 - {{'error': {{'message': 'Weekly usage limit reached', " @@ -21,6 +32,23 @@ def _quota_429(lane: str) -> Exception: return exc +def _plain_429(lane: str) -> Exception: + exc = Exception(f"Error code: 429 - {{'error': {{'message': 'Rate limit exceeded, retry in 20s', 'lane': '{lane}'}}}}") + exc.status_code = 429 + return exc + + +def _payment_402(lane: str) -> Exception: + exc = Exception(f"Error code: 402 - {{'error': {{'message': 'Insufficient credits', 'lane': '{lane}'}}}}") + exc.status_code = 402 + return exc + + +def _quarantine_remaining(base_url: str) -> float: + key = ac._unhealthy_cache_key("custom", base_url) + return ac._aux_unhealthy_until[key] - time.time() + + def _client(base_url: str, create): client = MagicMock() client.base_url = base_url @@ -47,7 +75,6 @@ def _walk_patches(primary, main_chain_selections): patch("agent.auxiliary_client._try_main_fallback_chain", side_effect=list(main_chain_selections) + [(None, None, "")]), patch("agent.auxiliary_client._try_payment_fallback", return_value=(None, None, "")), - patch("agent.auxiliary_client._mark_provider_unhealthy"), ) @@ -58,28 +85,29 @@ def test_sync_walk_advances_past_quota_limited_candidate_to_next_configured_entr lane_c = _client("http://127.0.0.1:3/v1", MagicMock(return_value=_ok_response("OK"))) patches = _walk_patches(primary, [(lane_b, "modelB", "custom"), (lane_c, "modelC", "custom")]) - with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5] as mark_unhealthy: + with patches[0], patches[1], patches[2], patches[3], patches[4]: result = call_llm(task="title_generation", messages=[{"role": "user", "content": "Reply OK"}]) assert result.choices[0].message.content == "OK" assert lane_b.chat.completions.create.call_count == 1 assert lane_c.chat.completions.create.call_count == 1 - # The quota-limited candidate is quarantined so the ordered re-walk skips it. - assert ("custom",) in {c.args for c in mark_unhealthy.call_args_list} - assert any(c.kwargs.get("base_url") == "http://127.0.0.1:2/v1" for c in mark_unhealthy.call_args_list) + # The quota-limited candidate is really quarantined (per endpoint) so the ordered re-walk skips it. + assert ac._is_provider_unhealthy("custom", "http://127.0.0.1:2/v1") + assert not ac._is_provider_unhealthy("custom", "http://127.0.0.1:3/v1") @pytest.mark.asyncio -async def test_async_walk_exhausts_every_configured_lane_once_then_raises_last_error(): - """T4: primary 429 → fallback[0] 429 → fallback[1] 429 → controlled exhaustion; each lane once.""" +async def test_async_walk_exhausts_every_configured_lane_once_then_raises_primary_error(): + """T4: primary 429 → fallback[0] 429 → fallback[1] 429 → controlled exhaustion; each lane once; + the primary's own error is what surfaces.""" primary = _client("http://127.0.0.1:1/v1", AsyncMock(side_effect=_quota_429("A"))) lane_b = _client("http://127.0.0.1:2/v1", AsyncMock(side_effect=_quota_429("B"))) lane_c = _client("http://127.0.0.1:3/v1", AsyncMock(side_effect=_quota_429("C"))) patches = _walk_patches(primary, [(lane_b, "modelB", "custom"), (lane_c, "modelC", "custom")]) - with patches[0], patches[1], patches[2], patches[3] as main_chain, patches[4] as discovery, patches[5], \ + with patches[0], patches[1], patches[2], patches[3] as main_chain, patches[4] as discovery, \ patch("agent.auxiliary_client._to_async_client", side_effect=lambda c, m, **kw: (c, m)): - with pytest.raises(Exception, match="usage_limit_reached"): + with pytest.raises(Exception, match="'lane': 'A'"): await async_call_llm(task="compression", messages=[{"role": "user", "content": "summarize"}]) assert lane_b.chat.completions.create.call_count == 1 @@ -88,3 +116,24 @@ async def test_async_walk_exhausts_every_configured_lane_once_then_raises_last_e assert main_chain.call_count == 3 # Exhaustion is controlled: discovery was consulted and found nothing, no fourth lane appended. assert discovery.call_count >= 1 + + +def test_candidate_quarantine_ttl_is_short_for_transient_429_and_long_for_payment(caplog): + """A per-minute 429 on a candidate hides the lane for seconds, not the 10-minute payment hold, + and the warning names the real class; a 402 candidate still gets the long hold.""" + caplog.set_level("WARNING", logger="agent.auxiliary_client") + primary = _client("http://127.0.0.1:1/v1", MagicMock(side_effect=_plain_429("A"))) + lane_b = _client("http://127.0.0.1:2/v1", MagicMock(side_effect=_plain_429("B"))) + lane_c = _client("http://127.0.0.1:3/v1", MagicMock(side_effect=_payment_402("C"))) + lane_d = _client("http://127.0.0.1:4/v1", MagicMock(return_value=_ok_response("OK"))) + + patches = _walk_patches( + primary, [(lane_b, "modelB", "custom"), (lane_c, "modelC", "custom"), (lane_d, "modelD", "custom")]) + with patches[0], patches[1], patches[2], patches[3], patches[4]: + result = call_llm(task="title_generation", messages=[{"role": "user", "content": "Reply OK"}]) + + assert result.choices[0].message.content == "OK" + assert 0 < _quarantine_remaining("http://127.0.0.1:2/v1") <= 60 + assert _quarantine_remaining("http://127.0.0.1:3/v1") > 60 + marks = [r.getMessage() for r in caplog.records if "marking local/custom unhealthy" in r.getMessage()] + assert marks and "payment" not in marks[0] and "rate limit" in marks[0]