hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(aux): quarantine a transient fallback-candidate failure for seconds, not the payment hold

Browse Source 
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).
This commit is contained in:
teknium1
2026-09-17 01:35:48 -07:00
committed by Teknium
parent 639fb57527
commit 951600dc79
4 changed files with 95 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
agent/auxiliary_client.py
View File

@@ -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

18
agent/auxiliary_health.py
View File

@@ -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 "")

9
tests/agent/test_auxiliary_client.py
View File

@@ -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

69
tests/agent/test_auxiliary_fallback_multihop.py
View File

@@ -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]