Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 91 additions & 3 deletions agent/credential_pool.py
Original file line number Diff line number Diff line change
Expand Up @@ -2544,6 +2544,58 @@ def peek(self) -> Optional[PooledCredential]:
available, _pending = self._available_entries()
return available[0] if available else None

def _adopt_rotated_entry_unlocked(
self,
entry: PooledCredential,
api_key_hint: Optional[str],
) -> Optional[PooledCredential]:
"""Return ``entry`` when it carries a usable token DIFFERENT from the
key that just failed; ``None`` when adopting it would change nothing.

The stale-key case: the agent was built with token T1, another
process/profile/keeper consumed the single-use refresh token and
wrote T2 to auth.json, and the pool (singleton-seeded entries resync
from the store) now holds T2. T1 gets a 401/403, matches no entry, and
the only "rotation" that can recover is *adopt T2*. Refreshing here
would be wrong twice over β€” it burns the fresh single-use refresh
token the other writers hold, and the entry is not the one that
failed.

Singleton-seeded xAI entries are resynced from auth.json first so an
in-process pool that has not touched the store since the rotation
still sees the newest pair. Nothing is marked exhausted; the caller's
unmatched-rotation streak bounds repeat visits.
"""
if not api_key_hint:
return None
candidate = entry
if self.provider == "xai-oauth":
try:
synced = self._sync_xai_oauth_entry_from_auth_store(entry)
except Exception: # pragma: no cover - defensive; sync is best-effort
synced = entry
if synced is not entry:
self._replace_entry(entry, synced)
self._persist()
candidate = synced
runtime_key = candidate.runtime_api_key
if not runtime_key or runtime_key == api_key_hint:
return None
if self._entry_needs_refresh(candidate):
# The lone entry's token is itself expiring; handing it back
# would just trade one dead key for another. Let the normal
# refresh/fallback path own it.
return None
logger.info(
"credential pool: failed %s key matches no entry but the pool's "
"only entry %s holds a different, unexpired token (rotated by "
"another process) β€” adopting it for one retry",
self.provider,
(candidate.label or candidate.id[:8]),
)
self._current_id = candidate.id
return candidate

def mark_exhausted_and_rotate(
self,
*,
Expand Down Expand Up @@ -2645,9 +2697,25 @@ def mark_exhausted_and_rotate(
next_entry, _pending = self._select_unlocked(refresh=False)
avail, _ = self._available_entries()
if next_entry is not None and len(avail) == 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
# A single-entry pool cannot rotate β€” but it CAN have
# been rotated underneath this agent. Single-use OAuth
# refresh (xai-oauth, openai-codex) means another
# process/profile/keeper re-mints the pair and the pool
# (via its auth.json sync) picks up the new token while
# the agent still dispatches with the key it was built
# with. That stale key is exactly the one that just
# failed and matches nothing. Handing back the lone
# entry then DOES change the credential, so try it once;
# the streak cap above bounds it (a second unmatched
# failure with the pool unchanged returns None β†’ fallback).
adopted = self._adopt_rotated_entry_unlocked(
next_entry, api_key_hint
)
if adopted is not None:
return adopted
# Same token the agent already holds: returning it would
# report a successful recovery without changing the
# credential, so the caller retries the same 401
# indefinitely. Let fallback/error propagation proceed.
self._unmatched_rotation_streak = 0
self._current_id = None
Expand Down Expand Up @@ -2810,6 +2878,26 @@ def try_refresh_matching(
),
None,
)
if (
entry is not None
and api_key_hint
and entry.runtime_api_key != api_key_hint
and not any(
candidate.runtime_api_key == api_key_hint
for candidate in self._entries
)
):
# The id-bound entry already holds a token other than the one
# that failed: another process/profile/keeper rotated the
# single-use pair underneath this agent. Force-refreshing
# would burn the fresh refresh token every other holder
# relies on, for a failure this token never had. Adopt the
# entry's current token instead; the caller retries with it.
adopted = self._adopt_rotated_entry_unlocked(
entry, api_key_hint
)
if adopted is not None:
return adopted
if entry is None:
if api_key_hint:
entry = next(
Expand Down
231 changes: 231 additions & 0 deletions tests/agent/test_credential_pool_adopt_rotated_entry.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,231 @@
"""Stale-agent-key recovery: adopt the pool's rotated token instead of
refreshing it or falling back.

Live incident 2026-09-19 03:16 PDT (``xai-oauth/grok-4.6`` β†’ fallback to
``claude-opus-5``): a long-lived gateway session was built with access token
T1. xAI refresh tokens are single-use; another process consumed the refresh
token and the single ``device_code`` pool entry now held T2 (the same instant
a cron session was completing grok calls with T2). T1 got
``HTTP 403 unauthenticated:bad-credentials``. Recovery attributed the failure
to ``agent.api_key`` (T1), which matched no entry, so:

* ``try_refresh_matching(api_key_hint=T1)`` β†’ ``None`` (nothing to refresh β€”
correct, refreshing would burn the fresh single-use token), then
* ``mark_exhausted_and_rotate`` hit the single-entry escape and returned
``None`` on the premise that handing back the lone entry "does not change
the credential" β€” false here, the entry holds T2 β€” so fallback fired while a
perfectly valid token sat in the pool.

The fix adopts the lone entry when its ``runtime_api_key`` differs from the
failed key and is not itself expiring. Bounded by the existing
unmatched-rotation streak (#70401) and the caller's refresh-attempt cap.
"""
import json

import pytest


def _seed_pool(tmp_path, monkeypatch, entries, provider="openrouter", store=None):
hermes_home = tmp_path / "hermes"
hermes_home.mkdir(parents=True, exist_ok=True)
payload = {"version": 1, "credential_pool": {provider: entries}}
if store is not None:
payload["providers"] = {provider: store}
(hermes_home / "auth.json").write_text(json.dumps(payload))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
from agent.credential_pool import load_pool

return load_pool(provider)


def _entry(idx, key, **extra):
base = {
"id": f"cred-{idx}",
"label": f"key-{idx}",
"auth_type": "api_key",
"priority": idx,
"source": "manual",
"access_token": key,
}
base.update(extra)
return base


class TestSingleEntryPoolAdoptsRotatedToken:

def test_unmatched_stale_key_adopts_lone_entry_once_then_bounds(
self, tmp_path, monkeypatch
):
"""Agent holds T1, pool's only entry holds T2 β†’ first rotation hands
back the entry (credential changes); a second unmatched failure with
the pool unchanged returns None so fallback can fire."""
pool = _seed_pool(tmp_path, monkeypatch, [_entry(0, "T2")])
assert pool.select() is not None

first = pool.mark_exhausted_and_rotate(
status_code=403,
error_context={"reason": "unauthorized"},
api_key_hint="T1",
)
assert first is not None
assert first.runtime_api_key == "T2"
assert pool.current() is not None and pool.current().id == first.id
# Nothing was quarantined β€” the failure belonged to a key the pool
# no longer holds.
assert all(e.last_status != "exhausted" for e in pool._entries)

# Caller retried with T1 again (it did not swap) β€” the streak cap
# (#70401) must stop the loop.
second = pool.mark_exhausted_and_rotate(
status_code=403,
error_context={"reason": "unauthorized"},
api_key_hint="T1",
)
assert second is None

def test_same_key_as_lone_entry_still_escapes(self, tmp_path, monkeypatch):
"""Regression guard for the original single-entry escape: when the
failed key IS the lone entry's key but was reported under an identity
that matches nothing, returning the entry changes nothing β†’ None."""
pool = _seed_pool(tmp_path, monkeypatch, [_entry(0, "T1")])
assert pool.select() is not None
# api_key_hint equals the entry's key but credential_id is bogus and
# the #79156 disagreement path drops the id β†’ treated as unmatched.
nxt = pool.mark_exhausted_and_rotate(
status_code=401,
error_context={"reason": "unauthorized"},
api_key_hint="T1",
credential_id="not-a-real-id",
)
# A real match on the key marks the entry exhausted and, with one
# entry, has nothing to rotate to.
assert nxt is None or nxt.runtime_api_key != "T1"

def test_unmatched_hint_with_no_hint_key_does_not_adopt(
self, tmp_path, monkeypatch
):
"""credential_id-only identity that matches nothing must keep the
old escape (no key to compare against β†’ no adoption)."""
pool = _seed_pool(tmp_path, monkeypatch, [_entry(0, "T2")])
assert pool.select() is not None
nxt = pool.mark_exhausted_and_rotate(
status_code=401,
error_context={"reason": "unauthorized"},
credential_id="ghost",
)
assert nxt is None

def test_expiring_lone_entry_is_not_adopted(self, tmp_path, monkeypatch):
"""Trading a dead key for an expiring one is not a recovery."""
pool = _seed_pool(tmp_path, monkeypatch, [_entry(0, "T2")])
assert pool.select() is not None
monkeypatch.setattr(pool, "_entry_needs_refresh", lambda entry: True)
nxt = pool.mark_exhausted_and_rotate(
status_code=401,
error_context={"reason": "unauthorized"},
api_key_hint="T1",
)
assert nxt is None


class TestTryRefreshMatchingAdoptsInsteadOfBurningRefreshToken:

def test_stale_id_bound_key_adopts_without_refresh(self, tmp_path, monkeypatch):
"""credential_id matches the entry but the entry already holds a
different token than the one that failed: adopt it, do NOT spend the
(single-use) refresh token."""
pool = _seed_pool(tmp_path, monkeypatch, [_entry(0, "T2")])
assert pool.select() is not None
calls = []

def _boom(entry, force=False):
calls.append(entry.id)
raise AssertionError("refresh must not run for a rotated entry")

monkeypatch.setattr(pool, "_refresh_entry", _boom)

got = pool.try_refresh_matching(api_key_hint="T1", credential_id="cred-0")
assert got is not None
assert got.runtime_api_key == "T2"
assert calls == []

def test_matching_key_still_refreshes(self, tmp_path, monkeypatch):
"""Regression guard: when the failed key IS the entry's key the
forced refresh path is unchanged."""
pool = _seed_pool(tmp_path, monkeypatch, [_entry(0, "T1")])
assert pool.select() is not None
calls = []

def _fake_refresh(entry, force=False):
calls.append((entry.id, force))
return entry

monkeypatch.setattr(pool, "_refresh_entry", _fake_refresh)
got = pool.try_refresh_matching(api_key_hint="T1", credential_id="cred-0")
assert got is not None
assert calls == [("cred-0", True)]

def test_hint_matches_other_entry_prefers_that_entry(self, tmp_path, monkeypatch):
"""Multi-entry: a stale id with a hint that matches ANOTHER entry must
refresh the key-matched entry (#79156 semantics), not adopt."""
pool = _seed_pool(
tmp_path, monkeypatch, [_entry(0, "T-a"), _entry(1, "T-b")]
)
assert pool.select() is not None
calls = []

def _fake_refresh(entry, force=False):
calls.append(entry.id)
return entry

monkeypatch.setattr(pool, "_refresh_entry", _fake_refresh)
# credential_id points at cred-0 but the failing key is cred-1's.
got = pool.try_refresh_matching(api_key_hint="T-b", credential_id="cred-0")
assert got is not None
# Existing behaviour: id wins when both are supplied; we only assert
# that adoption did not short-circuit the refresh.
assert calls, "refresh path was skipped for a key that IS in the pool"


class TestXaiSingletonResyncBeforeAdopt:

def test_xai_pool_resyncs_from_auth_store_before_adopting(
self, tmp_path, monkeypatch
):
"""Agent dispatched with T0; the in-process xai pool still holds the
later T1 but auth.json was rotated again to T2 by another
process/keeper: the adopt step must resync from the store and hand
back T2 (not the pool's own stale T1)."""
entry = _entry(
0,
"T1",
auth_type="oauth",
source="device_code",
refresh_token="R1",
last_refresh="2026-09-19T01:00:00Z",
)
store = {
"tokens": {"access_token": "T2", "refresh_token": "R2"},
"last_refresh": "2026-09-19T02:00:00Z",
}
pool = _seed_pool(
tmp_path, monkeypatch, [entry], provider="xai-oauth", store=store
)
# Pool loads may already resync; force the in-memory entry back to
# T1 to model a pool that has not touched the store since rotation.
from dataclasses import replace

with pool._lock:
cur = pool._entries[0]
pool._entries[0] = replace(cur, access_token="T1", refresh_token="R1")
monkeypatch.setattr(pool, "_entry_needs_refresh", lambda e: False)
assert pool.select() is not None

got = pool.mark_exhausted_and_rotate(
status_code=403,
error_context={"reason": "unauthorized"},
api_key_hint="T0",
)
assert got is not None
assert got.runtime_api_key == "T2"
assert got.refresh_token == "R2"
19 changes: 17 additions & 2 deletions tests/agent/test_credential_pool_routing.py
Original file line number Diff line number Diff line change
Expand Up @@ -495,8 +495,11 @@ def test_auth_refresh_targets_failing_key_not_pointer(self, tmp_path, monkeypatc
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."""
"""Legacy agents without a stable id: an unmatched key gets exactly
ONE adoption of the lone entry's (different) token β€” the pool may
have been rotated underneath the agent by another process (single-use
OAuth refresh; live 2026-09-19 xai-oauth incident) β€” and a second
unmatched failure stops without quarantining the healthy entry."""
pool = self._make_pool(
tmp_path, monkeypatch,
[self._entry(0, "pool-runtime-key")],
Expand All @@ -509,7 +512,19 @@ def test_unmatched_key_does_not_retry_only_pool_entry(
recovered, _ = recover_with_credential_pool(
agent, status_code=401, has_retried_429=False
)
# First pass: the pool's only entry holds a token the agent is NOT
# using β€” adopting it is a real credential change, so retry once.
assert recovered is True
agent._swap_credential.assert_called_once()
assert agent._swap_credential.call_args.args[0].id == "cred-0"
assert self._statuses(pool)["cred-0"] != "exhausted"

# Second pass with the agent STILL on the wrapper key (swap did not
# change what it dispatches with): bounded β€” stop, no quarantine.
agent._swap_credential.reset_mock()
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()
Expand Down
Loading