From 73c4b5a04511d26cabe9efdc87dd5ab5dc8dd87a Mon Sep 17 00:00:00 2001 From: Gille <4317663+helix4u@users.noreply.github.com> Date: Thu, 23 Jul 2026 15:54:00 -0600 Subject: [PATCH] fix(auth): stop stale-key credential recovery loops Track the selected credential by stable pool entry ID so token refreshes and shared cursor movement cannot detach failures from the entry that issued them. Stop unmatched single-entry pools from reporting a no-op rotation as successful recovery. Co-authored-by: Maxim Esipov --- agent/agent_init.py | 15 ++++ agent/agent_runtime_helpers.py | 74 +++++++++++----- agent/chat_completion_helpers.py | 12 +++ agent/credential_pool.py | 94 ++++++++++++++++----- run_agent.py | 1 + tests/agent/test_credential_pool_routing.py | 84 +++++++++++++++++- 6 files changed, 234 insertions(+), 46 deletions(-) diff --git a/agent/agent_init.py b/agent/agent_init.py index da266684767f..f2cf3e841f02 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -1341,6 +1341,21 @@ def init_agent( print("⚠️ Warning: API key appears invalid or missing") except Exception as e: raise RuntimeError(f"Failed to initialize OpenAI client: {e}") + + # Keep a stable identity for the pool entry that supplied this runtime. + # OAuth refreshes can replace the runtime token before a failed request is + # recovered, so the mutable API-key value alone cannot reliably attribute + # the failure to its source entry. + agent._credential_pool_entry_id = None + if agent._credential_pool is not None: + try: + agent._credential_pool_entry_id = ( + agent._credential_pool.entry_id_for_api_key( + getattr(agent, "api_key", None) + ) + ) + except Exception: + agent._credential_pool_entry_id = None # Provider fallback chain — ordered list of backup providers tried # when the primary is exhausted (rate-limit, overload, connection diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index f9ff6b112efd..e295a199be92 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -934,10 +934,30 @@ def recover_with_credential_pool( # failing entry exactly; fall back to current()'s key only when the agent # carries no key at all. _api_key_hint = getattr(agent, "api_key", None) or None + _raw_credential_id = getattr(agent, "_credential_pool_entry_id", None) + _credential_id = ( + _raw_credential_id + if isinstance(_raw_credential_id, str) and _raw_credential_id + else None + ) if not _api_key_hint: _cur = pool.current() if _cur: _api_key_hint = getattr(_cur, "runtime_api_key", None) + if not _credential_id: + _current_id = getattr(_cur, "id", None) + if isinstance(_current_id, str) and _current_id: + _credential_id = _current_id + + def _rotate_failed_credential(rotate_status: int): + kwargs = { + "status_code": rotate_status, + "error_context": error_context, + "api_key_hint": _api_key_hint, + } + if _credential_id: + kwargs["credential_id"] = _credential_id + return pool.mark_exhausted_and_rotate(**kwargs) effective_reason = classified_reason if effective_reason is None: @@ -972,11 +992,7 @@ def recover_with_credential_pool( # Runtime credentials can be resolved by a separate pool instance, # leaving this recovery pool without ``current_id``. Match the key # that actually failed instead of quarantining a different account. - next_entry = pool.mark_exhausted_and_rotate( - status_code=rotate_status, - error_context=error_context, - api_key_hint=_api_key_hint, - ) + next_entry = _rotate_failed_credential(rotate_status) if next_entry is not None: _ra().logger.info( "Credential %s (billing) — rotated to pool entry %s", @@ -995,8 +1011,13 @@ def recover_with_credential_pool( # Prefer the entry matching the failing key over the shared current() # pointer, for the same attribution reason as above. current_entry = None - if _api_key_hint: + if _credential_id: current_entry = next( + (e for e in pool.entries() if e.id == _credential_id), + None, + ) + if _api_key_hint: + current_entry = current_entry or next( (e for e in pool.entries() if e.runtime_api_key == _api_key_hint), None, ) @@ -1009,11 +1030,7 @@ def recover_with_credential_pool( current_last_status, ) rotate_status = status_code if status_code is not None else 429 - next_entry = pool.mark_exhausted_and_rotate( - status_code=rotate_status, - error_context=error_context, - api_key_hint=_api_key_hint, - ) + next_entry = _rotate_failed_credential(rotate_status) if next_entry is not None: _ra().logger.info( "Credential %s (rate limit, pre-exhausted) — rotated to pool entry %s", @@ -1037,11 +1054,7 @@ def recover_with_credential_pool( if not has_retried_429 and not usage_limit_reached: return False, True rotate_status = status_code if status_code is not None else 429 - next_entry = pool.mark_exhausted_and_rotate( - status_code=rotate_status, - error_context=error_context, - api_key_hint=_api_key_hint, - ) + next_entry = _rotate_failed_credential(rotate_status) if next_entry is not None: _ra().logger.info( "Credential %s (rate limit) — rotated to pool entry %s", @@ -1113,7 +1126,10 @@ def recover_with_credential_pool( # the shared pointer can reference a different, healthy entry, and # refreshing it would consume that entry's single-use refresh token # (or mark it exhausted on failure) for a failure it never had. - refreshed = pool.try_refresh_matching(api_key_hint=_api_key_hint) + refresh_kwargs = {"api_key_hint": _api_key_hint} + if _credential_id: + refresh_kwargs["credential_id"] = _credential_id + refreshed = pool.try_refresh_matching(**refresh_kwargs) if refreshed is not None: # ``try_refresh_matching()`` re-mints a fresh OAuth token and reports # success even when the upstream keeps rejecting it — a single-entry @@ -1145,11 +1161,7 @@ def recover_with_credential_pool( # Refresh failed — rotate to next credential instead of giving up. # The failed entry is already marked exhausted by the refresh attempt. rotate_status = status_code if status_code is not None else 401 - next_entry = pool.mark_exhausted_and_rotate( - status_code=rotate_status, - error_context=error_context, - api_key_hint=_api_key_hint, - ) + next_entry = _rotate_failed_credential(rotate_status) if next_entry is not None: _ra().logger.info( "Credential %s (auth refresh failed) — rotated to pool entry %s", @@ -1460,6 +1472,7 @@ def restore_primary_runtime(agent) -> bool: pool_matches_primary = False if pool is not None and pool_provider and not pool_matches_primary: agent._credential_pool = None + agent._credential_pool_entry_id = None try: from agent.credential_pool import load_pool @@ -1479,6 +1492,7 @@ def restore_primary_runtime(agent) -> bool: # the pool for its current best entry and swap the live credential in. # When the pool is absent, empty, or the entry has no usable key, we # keep the snapshot key (the existing behavior). Fixes #25205. + agent._credential_pool_entry_id = None pool = getattr(agent, "_credential_pool", None) if pool is not None and pool.has_available(): entry = pool.select() @@ -2075,6 +2089,9 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo # restore the original pool (issue #52727: pool reload is part of this # switch and must be reversible on rollback). _snapshot["_credential_pool"] = getattr(agent, "_credential_pool", _MISSING) + _snapshot["_credential_pool_entry_id"] = getattr( + agent, "_credential_pool_entry_id", _MISSING + ) try: # Clear the per-config context_length override so the new model's @@ -2131,6 +2148,7 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo # A pool bound to the old provider is worse than no pool: the # recovery guard rejects it and every later 401/429 skips rotation. agent._credential_pool = None + agent._credential_pool_entry_id = None try: from agent.credential_pool import load_pool agent._credential_pool = load_pool(new_provider) @@ -2140,7 +2158,6 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo "continuing without pool rotation this turn", new_provider, _pool_exc, ) - # ── Build new client ── if (new_provider or "").strip().lower() == "moa": from agent.moa_loop import build_moa_facade @@ -2236,6 +2253,17 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo reason="switch_model", shared=True, ) + + pool = getattr(agent, "_credential_pool", None) + if pool is not None: + try: + agent._credential_pool_entry_id = pool.entry_id_for_api_key( + getattr(agent, "api_key", None) + ) + except Exception: + agent._credential_pool_entry_id = None + else: + agent._credential_pool_entry_id = None except Exception: # Rollback every mutated field to the pre-swap snapshot so the agent # is left consistent (old model + old provider + old client) and the diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 37113b61acf0..f22036ee93fd 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -1739,6 +1739,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool fb_provider, fb_model, _pool_provider, ) agent._credential_pool = None + agent._credential_pool_entry_id = None if getattr(agent, "_credential_pool", None) is None: try: from agent.credential_pool import load_pool @@ -1801,6 +1802,17 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool # not only after a later credential-rotation rebuild. agent._replace_primary_openai_client(reason="fallback_timeout_apply") + fallback_pool = getattr(agent, "_credential_pool", None) + if fallback_pool is not None: + try: + agent._credential_pool_entry_id = ( + fallback_pool.entry_id_for_api_key(agent.api_key) + ) + except Exception: + agent._credential_pool_entry_id = None + else: + agent._credential_pool_entry_id = None + # Re-evaluate prompt caching for the new provider/model agent._use_prompt_caching, agent._use_native_cache_layout = ( agent._anthropic_prompt_cache_policy( diff --git a/agent/credential_pool.py b/agent/credential_pool.py index d5d652ad7414..516e7f61db61 100644 --- a/agent/credential_pool.py +++ b/agent/credential_pool.py @@ -622,6 +622,28 @@ class CredentialPool: with self._lock: return self._current_unlocked() + def entry_id_for_api_key(self, api_key_hint: Any = None) -> Optional[str]: + """Return the stable id for the runtime credential in use. + + Prefer the current selection when it still supplies ``api_key_hint``. + If the cursor was cleared, fall back to an unambiguous key match. + """ + with self._lock: + current = self._current_unlocked() + if current is not None and ( + api_key_hint is None + or current.runtime_api_key == api_key_hint + ): + return current.id + if api_key_hint is None: + return None + matches = [ + entry + for entry in self._entries + if entry.runtime_api_key == api_key_hint + ] + return matches[0].id if len(matches) == 1 else None + def _replace_entry(self, old: PooledCredential, new: PooledCredential) -> None: """Swap an entry in-place by id, preserving sort order.""" for idx, entry in enumerate(self._entries): @@ -1763,10 +1785,17 @@ class CredentialPool: status_code: Optional[int], error_context: Optional[Dict[str, Any]] = None, api_key_hint: Optional[str] = None, + credential_id: Optional[str] = None, ) -> Optional[PooledCredential]: with self._lock: entry = None - if api_key_hint: + identity_supplied = bool(credential_id or api_key_hint) + if credential_id: + entry = next( + (e for e in self._entries if e.id == credential_id), + None, + ) + if entry is None and api_key_hint: # Prefer the specific entry whose API key matches the one that # actually failed. When this pool was freshly loaded from disk # (another process already rotated), current() is None and @@ -1775,20 +1804,26 @@ class CredentialPool: (e for e in self._entries if e.runtime_api_key == api_key_hint), None, ) - if entry is None: - # The failed key is identifiable but matches no entry - # (rotated away, or a wrapper whose runtime key differs). - # Falling through to current()/_select_unlocked() would - # mark an INNOCENT healthy key exhausted for the full - # cooldown TTL. Don't guess — just hand back a fresh - # selection so the caller can retry. - logger.info( - "credential pool: failed key hint matched no %s entry; " - "rotating without marking any credential exhausted", - self.provider, - ) + if entry is None and identity_supplied: + # The failed credential is identifiable but matches no entry + # (rotated away, or a wrapper whose runtime key differs). + # Falling through to current()/_select_unlocked() would mark an + # innocent healthy key exhausted for the full cooldown TTL. + logger.info( + "credential pool: failed credential identity matched no %s " + "entry; rotating without marking any credential exhausted", + self.provider, + ) + self._current_id = None + next_entry = self._select_unlocked() + if next_entry is not None and len(self._available_entries()) == 1: + # A single-entry pool cannot rotate. Returning its only + # entry reports a successful recovery without changing + # the credential, so the caller retries the same 401 + # indefinitely. Let fallback/error propagation proceed. self._current_id = None - return self._select_unlocked() + return None + return next_entry if entry is None: entry = self._current_unlocked() or self._select_unlocked() if entry is None: @@ -1806,12 +1841,13 @@ class CredentialPool: # disconnects (a ~2.5min hang with no error surfaced to the user). # Mark every entry sharing the failed key so the pool can reach the # "no available entries" state and let the error propagate. - if api_key_hint: + failed_runtime_key = getattr(entry, "runtime_api_key", None) + if identity_supplied and failed_runtime_key: siblings_marked = False for sibling in self._entries: if sibling.id == entry.id: continue - if sibling.runtime_api_key == api_key_hint: + if sibling.runtime_api_key == failed_runtime_key: self._mark_exhausted( sibling, status_code, error_context, persist=False ) @@ -1885,9 +1921,11 @@ class CredentialPool: return self._try_refresh_current_unlocked() def try_refresh_matching( - self, api_key_hint: Optional[str] = None + self, + api_key_hint: Optional[str] = None, + credential_id: Optional[str] = None, ) -> Optional[PooledCredential]: - """Force-refresh the entry that supplied ``api_key_hint``. + """Force-refresh the entry that supplied the failed request. Direct provider integrations may reload the pool after a request has already failed, so they cannot rely on ``current_id`` identifying the @@ -1897,17 +1935,29 @@ class CredentialPool: """ with self._lock: entry = None - if api_key_hint: + if credential_id: entry = next( ( candidate for candidate in self._entries - if candidate.runtime_api_key == api_key_hint + if candidate.id == credential_id ), None, ) - else: - entry = self._current_unlocked() or self._select_unlocked(refresh=False) + if entry is None: + if api_key_hint: + entry = next( + ( + candidate + for candidate in self._entries + if candidate.runtime_api_key == api_key_hint + ), + None, + ) + else: + entry = self._current_unlocked() or self._select_unlocked( + refresh=False + ) if entry is None: return None self._current_id = entry.id diff --git a/run_agent.py b/run_agent.py index 75846d5bec81..e4cd9f54e8ce 100644 --- a/run_agent.py +++ b/run_agent.py @@ -4963,6 +4963,7 @@ class AIAgent: def _swap_credential(self, entry) -> None: runtime_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "") runtime_base = getattr(entry, "runtime_base_url", None) or getattr(entry, "base_url", None) or self.base_url + self._credential_pool_entry_id = getattr(entry, "id", None) from hermes_cli.route_identity import normalize_route_base_url route_changed = normalize_route_base_url(self.base_url) != normalize_route_base_url( diff --git a/tests/agent/test_credential_pool_routing.py b/tests/agent/test_credential_pool_routing.py index 8f093f86b0e1..55b907a717e5 100644 --- a/tests/agent/test_credential_pool_routing.py +++ b/tests/agent/test_credential_pool_routing.py @@ -405,11 +405,12 @@ class TestFailureAttribution: entry.update(overrides) return entry - def _agent(self, pool, failing_key): + def _agent(self, pool, failing_key, credential_id=None): return SimpleNamespace( provider="anthropic", api_key=failing_key, _credential_pool=pool, + _credential_pool_entry_id=credential_id, _swap_credential=MagicMock(), ) @@ -521,3 +522,84 @@ class TestFailureAttribution: assert statuses["cred-0"] != "exhausted" swapped = agent._swap_credential.call_args[0][0] assert swapped.id == "cred-0" + + def test_auth_refresh_uses_stable_id_after_runtime_key_changes( + self, tmp_path, monkeypatch + ): + """A refreshed pool token must not detach the failed request from the + entry that supplied its now-stale runtime key.""" + pool = self._make_pool( + tmp_path, monkeypatch, + [self._entry(0, "new-runtime-key")], + ) + selected = pool.select() + assert selected.id == "cred-0" + assert pool.entry_id_for_api_key("new-runtime-key") == "cred-0" + + agent = self._agent( + pool, + failing_key="stale-runtime-key", + credential_id="cred-0", + ) + agent._is_entitlement_failure = MagicMock(return_value=False) + + from agent.agent_runtime_helpers import recover_with_credential_pool + + recovered, _ = recover_with_credential_pool( + agent, status_code=401, has_retried_429=False + ) + + assert recovered is False + assert self._statuses(pool)["cred-0"] == "exhausted" + agent._swap_credential.assert_not_called() + + def test_unmatched_key_does_not_retry_only_pool_entry( + self, tmp_path, monkeypatch + ): + """Legacy agents without a stable id must stop when an unmatched key + has no different credential to rotate to.""" + pool = self._make_pool( + tmp_path, monkeypatch, + [self._entry(0, "pool-runtime-key")], + ) + agent = self._agent(pool, failing_key="wrapper-runtime-key") + agent._is_entitlement_failure = MagicMock(return_value=False) + + from agent.agent_runtime_helpers import recover_with_credential_pool + + recovered, _ = recover_with_credential_pool( + agent, status_code=401, has_retried_429=False + ) + + assert recovered is False + assert self._statuses(pool)["cred-0"] != "exhausted" + agent._swap_credential.assert_not_called() + + def test_stable_id_rotates_from_failed_entry_when_cursor_points_elsewhere( + self, tmp_path, monkeypatch + ): + """Stable identity wins over both a stale key and the shared cursor.""" + pool = self._make_pool( + tmp_path, monkeypatch, + [self._entry(0, "key-a"), self._entry(1, "key-b-new")], + ) + assert pool.select().id == "cred-0" + agent = self._agent( + pool, + failing_key="key-b-old", + credential_id="cred-1", + ) + agent._is_entitlement_failure = MagicMock(return_value=False) + + from agent.agent_runtime_helpers import recover_with_credential_pool + + recovered, _ = recover_with_credential_pool( + agent, status_code=401, has_retried_429=False + ) + + assert recovered is True + statuses = self._statuses(pool) + assert statuses["cred-1"] == "exhausted" + assert statuses["cred-0"] != "exhausted" + swapped = agent._swap_credential.call_args[0][0] + assert swapped.id == "cred-0"