fix(anthropic): serialize Claude Code OAuth refresh under the auth-store lock

agent/anthropic_adapter.py

@@ -1112,18 +1112,64 @@ def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]:
         logger.debug("No refresh token available — cannot refresh")
         return None
 
+    # Claude Code OAuth refresh tokens are single-use. Multiple concurrent
+    # callers (every anthropic_messages API call runs
+    # _try_refresh_anthropic_client_credentials, plus auxiliary tasks, plus
+    # the credential pool) can otherwise each POST the same refresh token,
+    # invalidate one another, and drive Anthropic's refresh endpoint into
+    # 429s — after which the sole Anthropic credential is marked exhausted
+    # and every live turn silently falls back to another provider.
+    #
+    # Serialize the read→POST→write-back through the shared cross-process
+    # auth-store flock (reentrant, so the credential-pool refresh path that
+    # already holds it does not deadlock). Once inside the lock, re-read the
+    # live credential file: if the winner already rotated the token, adopt it
+    # and skip the POST entirely.
     try:
-        refreshed = refresh_anthropic_oauth_pure(refresh_token, use_json=False)
-        _write_claude_code_credentials(
-            refreshed["access_token"],
-            refreshed["refresh_token"],
-            refreshed["expires_at_ms"],
-        )
-        logger.debug("Successfully refreshed Claude Code OAuth token")
-        return refreshed["access_token"]
-    except Exception as e:
-        logger.debug("Failed to refresh Claude Code token: %s", e)
-        return None
+        from hermes_cli.auth import _auth_store_lock, AUTH_LOCK_TIMEOUT_SECONDS
+        from utils import env_float
+
+        refresh_timeout_seconds = env_float("HERMES_ANTHROPIC_REFRESH_TIMEOUT_SECONDS", 30)
+        lock_timeout = max(float(AUTH_LOCK_TIMEOUT_SECONDS), float(refresh_timeout_seconds) + 5.0)
+        _lock_cm = _auth_store_lock(timeout_seconds=lock_timeout)
+    except Exception:
+        # If the lock helper is unavailable for any reason, fall back to the
+        # unserialized path rather than failing the refresh outright.
+        _lock_cm = None
+
+    def _do_refresh() -> Optional[str]:
+        # Re-read inside the lock: the winner may have already rotated the
+        # token while we waited, in which case adopt it and skip the POST.
+        latest = read_claude_code_credentials()
+        if latest:
+            latest_token = latest.get("accessToken", "")
+            latest_exp = latest.get("expiresAt", 0) or 0
+            if (
+                latest_token
+                and latest_token != creds.get("accessToken", "")
+                and latest_exp > 0
+                and is_claude_code_token_valid(latest)
+            ):
+                logger.debug("Adopted concurrently-refreshed Claude Code OAuth token (in-lock)")
+                return latest_token
+        effective_refresh = (latest or {}).get("refreshToken", "") or refresh_token
+        try:
+            refreshed = refresh_anthropic_oauth_pure(effective_refresh, use_json=False)
+            _write_claude_code_credentials(
+                refreshed["access_token"],
+                refreshed["refresh_token"],
+                refreshed["expires_at_ms"],
+            )
+            logger.debug("Successfully refreshed Claude Code OAuth token")
+            return refreshed["access_token"]
+        except Exception as e:
+            logger.debug("Failed to refresh Claude Code token: %s", e)
+            return None
+
+    if _lock_cm is None:
+        return _do_refresh()
+    with _lock_cm:
+        return _do_refresh()
 
 
 def _write_claude_code_credentials(

agent/credential_pool.py

@@ -113,6 +113,11 @@ def _load_config_safe() -> Optional[dict]:
 EXHAUSTED_TTL_401_SECONDS = 5 * 60           # 5 minutes
 EXHAUSTED_TTL_429_SECONDS = 60 * 60          # 1 hour
 EXHAUSTED_TTL_DEFAULT_SECONDS = 60 * 60      # 1 hour
+# Anthropic Claude Code OAuth token-refresh 429s are usually short-lived
+# control-plane throttles, not a model-usage quota window.  Treating them like
+# generic provider 429s freezes Claude for an hour and pushes every live turn to
+# fallback even though retrying shortly afterwards often succeeds.
+ANTHROPIC_OAUTH_REFRESH_429_TTL_SECONDS = 60  # 1 minute
 
 # Pool key prefix for custom OpenAI-compatible endpoints.
 # Custom endpoints all share provider='custom' but are keyed by their
@@ -965,14 +970,30 @@ def _refresh_entry(self, entry: PooledCredential, *, force: bool) -> Optional[Po
                 self._mark_exhausted(entry, None)
             return None
 
-        # Codex OAuth refresh tokens are single-use.  The sync→POST→write-back
+        # OAuth refresh tokens are single-use.  The sync→POST→write-back
         # sequence below must run atomically across Hermes processes: otherwise
         # two processes can both adopt the same on-disk token, both POST it, and
-        # the loser gets ``refresh_token_reused``.  Serialize the whole sequence
-        # through the shared cross-process auth-store flock (the same lock and
-        # extended-timeout pattern used by resolve_codex_runtime_credentials()).
-        # When a waiter finally acquires the lock, the in-lock re-sync below
-        # picks up the rotated token the winner persisted and skips the POST.
+        # the loser gets refresh_token_reused / invalid_grant or upstream refresh
+        # throttling.  Serialize refresh paths that share a singleton backing
+        # store through the shared cross-process auth-store flock.  When a waiter
+        # finally acquires the lock, the in-lock re-sync picks up the rotated
+        # token the winner persisted and can skip the POST.
+        if self.provider == "anthropic" and entry.source == "claude_code":
+            refresh_timeout_seconds = auth_mod.env_float(
+                "HERMES_ANTHROPIC_REFRESH_TIMEOUT_SECONDS", 30
+            )
+            lock_timeout = max(
+                float(auth_mod.AUTH_LOCK_TIMEOUT_SECONDS),
+                float(refresh_timeout_seconds) + 5.0,
+            )
+            with _auth_store_lock(timeout_seconds=lock_timeout):
+                synced = self._sync_anthropic_entry_from_credentials_file(entry)
+                if synced is not entry:
+                    entry = synced
+                    if not force and not self._entry_needs_refresh(entry):
+                        return entry
+                return self._refresh_entry_impl(entry, force=force)
+
         if self.provider == "openai-codex":
             refresh_timeout_seconds = auth_mod.env_float(
                 "HERMES_CODEX_REFRESH_TIMEOUT_SECONDS", 20
@@ -1069,6 +1090,22 @@ def _refresh_entry_impl(
                 return entry
         except Exception as exc:
             logger.debug("Credential refresh failed for %s/%s: %s", self.provider, entry.id, exc)
+            refresh_status_code = getattr(exc, "code", None)
+            try:
+                refresh_status_code = int(refresh_status_code) if refresh_status_code is not None else None
+            except (TypeError, ValueError):
+                refresh_status_code = None
+            refresh_error_context: Optional[Dict[str, Any]] = None
+            if (
+                self.provider == "anthropic"
+                and entry.source == "claude_code"
+                and refresh_status_code == 429
+            ):
+                refresh_error_context = {
+                    "reason": "oauth_refresh_rate_limited",
+                    "message": "Anthropic OAuth token refresh was rate limited; using a short retry cooldown",
+                    "reset_at": time.time() + ANTHROPIC_OAUTH_REFRESH_429_TTL_SECONDS,
+                }
             # For anthropic claude_code entries: the refresh token may have been
             # consumed by another process. Check if ~/.claude/.credentials.json
             # has a newer token pair and retry once.
@@ -1318,7 +1355,7 @@ def _refresh_entry_impl(
                         self._current_id = None
                     self._persist(removed_ids=removed_ids)
                     return None
-            self._mark_exhausted(entry, None)
+            self._mark_exhausted(entry, refresh_status_code, refresh_error_context)
             return None
 
         updated = replace(

tests/agent/test_anthropic_adapter.py

@@ -568,6 +568,96 @@ def test_failed_refresh_returns_none(self, tmp_path, monkeypatch):
         with patch("urllib.request.urlopen", side_effect=Exception("network error")):
             assert _refresh_oauth_token(creds) is None
 
+    def test_refresh_runs_under_auth_store_lock(self, tmp_path, monkeypatch):
+        """The single-use refresh POST must be serialized by the shared auth lock.
+
+        Concurrent anthropic_messages turns / aux tasks / the pool all funnel
+        through _refresh_oauth_token; without the lock they each replay the same
+        single-use refresh token and trigger 429s that exhaust the sole
+        Anthropic credential and force a provider fallback.
+        """
+        monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
+
+        lock_calls = []
+        import contextlib
+
+        @contextlib.contextmanager
+        def _fake_lock(timeout_seconds=None):
+            lock_calls.append(timeout_seconds)
+            yield
+
+        monkeypatch.setattr("hermes_cli.auth._auth_store_lock", _fake_lock)
+
+        creds = {
+            "accessToken": "old-token",
+            "refreshToken": "refresh-123",
+            "expiresAt": int(time.time() * 1000) - 3600_000,
+        }
+        # No concurrently-rotated token on disk → the POST path runs (under lock).
+        monkeypatch.setattr("agent.anthropic_adapter.read_claude_code_credentials", lambda: None)
+        monkeypatch.setattr(
+            "agent.anthropic_adapter.refresh_anthropic_oauth_pure",
+            lambda refresh_token, use_json=False: {
+                "access_token": "new-token-abc",
+                "refresh_token": "new-refresh-456",
+                "expires_at_ms": int(time.time() * 1000) + 7200_000,
+            },
+        )
+        monkeypatch.setattr("agent.anthropic_adapter._write_claude_code_credentials", lambda *a, **k: None)
+
+        result = _refresh_oauth_token(creds)
+
+        assert result == "new-token-abc"
+        assert lock_calls, "refresh POST must be wrapped in the shared auth-store lock"
+
+    def test_refresh_adopts_concurrently_rotated_token_without_posting(self, tmp_path, monkeypatch):
+        """A lock waiter must adopt the winner's rotated token, not re-POST.
+
+        Simulates process B entering _refresh_oauth_token after process A has
+        already refreshed and written ~/.claude/.credentials.json. B must return
+        the fresh token and never call refresh_anthropic_oauth_pure again.
+        """
+        monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
+
+        import contextlib
+
+        @contextlib.contextmanager
+        def _fake_lock(timeout_seconds=None):
+            yield
+
+        monkeypatch.setattr("hermes_cli.auth._auth_store_lock", _fake_lock)
+
+        stale_creds = {
+            "accessToken": "stale-access",
+            "refreshToken": "stale-refresh",
+            "expiresAt": int(time.time() * 1000) - 3600_000,
+        }
+        # First read (pre-lock adoption check): still stale.
+        # Second read (in-lock): winner has rotated a fresh valid token.
+        reads = [
+            dict(stale_creds),
+            {
+                "accessToken": "winner-access",
+                "refreshToken": "winner-refresh",
+                "expiresAt": int(time.time() * 1000) + 7200_000,
+            },
+        ]
+
+        def _fake_read():
+            return reads.pop(0) if reads else reads_last[0]
+
+        reads_last = [reads[-1]]
+        monkeypatch.setattr("agent.anthropic_adapter.read_claude_code_credentials", _fake_read)
+
+        def _should_not_post(*args, **kwargs):
+            raise AssertionError("must adopt rotated token instead of POSTing the consumed refresh token")
+
+        monkeypatch.setattr("agent.anthropic_adapter.refresh_anthropic_oauth_pure", _should_not_post)
+
+        result = _refresh_oauth_token(stale_creds)
+
+        assert result == "winner-access"
+
 
 class TestWriteClaudeCodeCredentials:
     def test_writes_new_file(self, tmp_path, monkeypatch):