fix(auth): make xAI OAuth pools multi-account resilient

agent/agent_runtime_helpers.py

@@ -837,7 +837,14 @@ def recover_with_credential_pool(
 
     if effective_reason == FailoverReason.billing:
         rotate_status = status_code if status_code is not None else 402
-        next_entry = pool.mark_exhausted_and_rotate(status_code=rotate_status, error_context=error_context)
+        next_entry = pool.mark_exhausted_and_rotate(
+            status_code=rotate_status,
+            error_context=error_context,
+            # 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.
+            api_key_hint=getattr(agent, "api_key", None),
+        )
         if next_entry is not None:
             _ra().logger.info(
                 "Credential %s (billing) — rotated to pool entry %s",
@@ -3134,6 +3141,10 @@ def extract_api_error_context(error: Exception) -> Dict[str, Any]:
         if isinstance(reason, str) and reason.strip():
             context["reason"] = reason.strip()
         message = payload.get("message") or payload.get("error_description")
+        if not message and isinstance(payload.get("error"), str):
+            # xAI uses a top-level string ``error`` beside a structured
+            # ``code`` (for example personal-team-blocked:spending-limit).
+            message = payload.get("error")
         if isinstance(message, str) and message.strip():
             context["message"] = message.strip()
         for key in ("resets_at", "reset_at"):

agent/credential_pool.py

@@ -824,6 +824,45 @@ def _sync_xai_oauth_entry_from_auth_store(self, entry: PooledCredential) -> Pool
             logger.debug("Failed to sync xAI OAuth entry from auth.json: %s", exc)
         return entry
 
+    def _sync_xai_oauth_entry_from_pool_store(
+        self, entry: PooledCredential
+    ) -> PooledCredential:
+        """Adopt a token pair rotated by another pool instance.
+
+        Direct xAI integrations load a fresh ``CredentialPool`` for each
+        request. Their in-memory locks therefore cannot protect xAI's
+        single-use refresh token across concurrent requests or processes.
+        This helper is called while the shared auth-store lock is held and
+        re-reads the exact persisted row before a refresh POST is attempted.
+        """
+        if self.provider != "xai-oauth":
+            return entry
+        try:
+            persisted = next(
+                (
+                    payload
+                    for payload in read_credential_pool(self.provider)
+                    if isinstance(payload, dict) and payload.get("id") == entry.id
+                ),
+                None,
+            )
+            if not isinstance(persisted, dict):
+                return entry
+            stored = PooledCredential.from_dict(self.provider, persisted)
+            if (
+                stored.access_token != entry.access_token
+                or stored.refresh_token != entry.refresh_token
+            ):
+                logger.debug(
+                    "Pool entry %s: adopting xAI OAuth tokens rotated by another pool instance",
+                    entry.id,
+                )
+                self._replace_entry(entry, stored)
+                return stored
+        except Exception as exc:
+            logger.debug("Failed to sync xAI OAuth entry from credential pool: %s", exc)
+        return entry
+
     def _sync_nous_entry_from_auth_store(self, entry: PooledCredential) -> PooledCredential:
         """Sync a Nous pool entry from auth.json if tokens differ.
 
@@ -1040,6 +1079,22 @@ def _refresh_entry(self, entry: PooledCredential, *, force: bool) -> Optional[Po
                     if not force and not self._entry_needs_refresh(entry):
                         return entry
                 return self._refresh_entry_impl(entry, force=force)
+        if self.provider == "xai-oauth":
+            refresh_timeout_seconds = auth_mod.env_float(
+                "HERMES_XAI_REFRESH_TIMEOUT_SECONDS", 20
+            )
+            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_xai_oauth_entry_from_pool_store(entry)
+                if (
+                    synced.access_token != entry.access_token
+                    or synced.refresh_token != entry.refresh_token
+                ):
+                    return synced
+                return self._refresh_entry_impl(synced, force=force)
         return self._refresh_entry_impl(entry, force=force)
 
     def _refresh_entry_impl(
@@ -1538,8 +1593,8 @@ def _available_entries(self, *, clear_expired: bool = False, refresh: bool = Fal
             self._persist(removed_ids=entries_to_prune)
         return available
 
-    def _select_unlocked(self) -> Optional[PooledCredential]:
-        available = self._available_entries(clear_expired=True, refresh=True)
+    def _select_unlocked(self, *, refresh: bool = True) -> Optional[PooledCredential]:
+        available = self._available_entries(clear_expired=True, refresh=refresh)
         if not available:
             self._current_id = None
             logger.info("credential pool: no available entries (all exhausted or empty)")
@@ -1668,6 +1723,35 @@ def try_refresh_current(self) -> Optional[PooledCredential]:
         with self._lock:
             return self._try_refresh_current_unlocked()
 
+    def try_refresh_matching(
+        self, api_key_hint: Optional[str] = None
+    ) -> Optional[PooledCredential]:
+        """Force-refresh the entry that supplied ``api_key_hint``.
+
+        Direct provider integrations may reload the pool after a request has
+        already failed, so they cannot rely on ``current_id`` identifying the
+        issuing credential. With no hint, select an entry without first doing
+        the normal proactive refresh; the forced refresh below must consume a
+        rotating refresh token exactly once.
+        """
+        with self._lock:
+            entry = 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() or self._select_unlocked(refresh=False)
+            if entry is None:
+                return None
+            self._current_id = entry.id
+            return self._try_refresh_current_unlocked()
+
     def _try_refresh_current_unlocked(self) -> Optional[PooledCredential]:
         entry = self.current()
         if entry is None:

agent/error_classifier.py

@@ -123,6 +123,25 @@ def is_auth(self) -> bool:
     "not available on the free tier",
 ]
 
+# xAI's explicit Grok credit-exhaustion code. Keep the HTTP 403 special case
+# provider-scoped: other providers' generic billing codes historically remain
+# auth failures when they arrive as 403.
+_XAI_SPENDING_LIMIT_ERROR_CODE = "personal-team-blocked:spending-limit"
+
+# Structured provider codes that mean the account cannot serve paid traffic
+# until credits/subscription capacity is restored. xAI returns its explicit
+# Grok spending-limit signal as HTTP 403 rather than 402.
+_BILLING_ERROR_CODES = frozenset({
+    "insufficient_quota",
+    "billing_not_active",
+    "payment_required",
+    "insufficient_credits",
+    "no_usable_credits",
+    "balance_depleted",
+    "model_not_supported_on_free_tier",
+    _XAI_SPENDING_LIMIT_ERROR_CODE,
+})
+
 # Patterns that indicate rate limiting (transient, will resolve)
 _RATE_LIMIT_PATTERNS = [
     "rate limit",
@@ -906,7 +925,11 @@ def _classify_by_status(
         # OpenRouter 403 "key limit exceeded" is actually billing. Other
         # providers also use 403 for account-plan or credit exhaustion.
         if (
-            "key limit exceeded" in error_msg
+            (
+                provider == "xai-oauth"
+                and error_code.lower() == _XAI_SPENDING_LIMIT_ERROR_CODE
+            )
+            or "key limit exceeded" in error_msg
             or "spending limit" in error_msg
             or any(p in error_msg for p in _BILLING_PATTERNS)
         ):
@@ -1292,15 +1315,7 @@ def _classify_by_error_code(
             should_rotate_credential=True,
         )
 
-    if code_lower in {
-        "insufficient_quota",
-        "billing_not_active",
-        "payment_required",
-        "insufficient_credits",
-        "no_usable_credits",
-        "balance_depleted",
-        "model_not_supported_on_free_tier",
-    }:
+    if code_lower in _BILLING_ERROR_CODES:
         return result_fn(
             FailoverReason.billing,
             retryable=False,

hermes_cli/auth_commands.py

@@ -314,8 +314,9 @@ def auth_add_command(args) -> None:
             _oauth_default_label(provider, len(pool.entries()) + 1),
         )
         # Add a distinct, self-contained pool entry per account (matching the
-        # xai-oauth / qwen-oauth patterns) instead of
-        # routing through the singleton ``_save_codex_tokens`` save path.
+        # qwen-oauth / minimax-oauth multi-account patterns, and the
+        # xai-oauth path below) instead of routing through the singleton
+        # ``_save_codex_tokens`` save path.
         # The singleton round-trip collapsed every added account into the
         # latest login: a second ``hermes auth add openai-codex`` overwrote
         # the first account's singleton-mirrored ``device_code`` entry rather
@@ -349,19 +350,40 @@ def auth_add_command(args) -> None:
             timeout_seconds=getattr(args, "timeout", None) or 20.0,
             open_browser=not getattr(args, "no_browser", False),
         )
-        auth_mod._save_xai_oauth_tokens(
-            creds["tokens"],
-            discovery=creds.get("discovery"),
-            redirect_uri=creds.get("redirect_uri", ""),
-            last_refresh=creds.get("last_refresh"),
-            auth_mode="oauth_device_code",
+        label = (getattr(args, "label", None) or "").strip() or label_from_token(
+            creds["tokens"]["access_token"],
+            _oauth_default_label(provider, len(pool.entries()) + 1),
         )
-        pool = load_pool(provider)
-        entry = next((e for e in pool.entries() if getattr(e, "source", "") == "device_code"), None)
-        shown_label = entry.label if entry is not None else label_from_token(
-            creds["tokens"]["access_token"], _oauth_default_label(provider, 1)
+        # Add a distinct, self-contained pool entry per account (matching the
+        # openai-codex / qwen-oauth / minimax-oauth patterns) instead of
+        # routing through the singleton ``_save_xai_oauth_tokens`` save path.
+        # The singleton round-trip collapsed every added account into the
+        # latest login: a second ``hermes auth add xai-oauth`` overwrote the
+        # first account's singleton-mirrored ``device_code`` entry rather than
+        # creating an independent one. ``manual:device_code`` entries refresh
+        # from their own token pair (``_sync_xai_oauth_entry_from_auth_store``
+        # only adopts the singleton for ``source=="device_code"``), so they
+        # need no singleton shadow.
+        entry = PooledCredential(
+            provider=provider,
+            id=uuid.uuid4().hex[:6],
+            label=label,
+            auth_type=AUTH_TYPE_OAUTH,
+            priority=0,
+            source=SOURCE_MANUAL_DEVICE_CODE,
+            access_token=creds["tokens"]["access_token"],
+            refresh_token=creds["tokens"].get("refresh_token"),
+            base_url=creds.get("base_url") or auth_mod.DEFAULT_XAI_OAUTH_BASE_URL,
+            last_refresh=creds.get("last_refresh"),
         )
-        print(f'Saved {provider} OAuth credentials: "{shown_label}"')
+        first_credential = not pool.entries()
+        pool.add_entry(entry)
+        # Adding the first xAI credential should make it the active provider
+        # (the old singleton save path did this implicitly via
+        # _save_provider_state). Subsequent adds leave the active provider as-is.
+        if first_credential:
+            auth_mod.mark_provider_active_if_unset(provider)
+        print(f'Added {provider} OAuth credential #{len(pool.entries())}: "{entry.label}"')
         return
 
     if provider == "qwen-oauth":

plugins/web/xai/provider.py

@@ -270,7 +270,10 @@ def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
                         "refresh and retrying once.",
                     )
                     try:
-                        refreshed = resolve_xai_http_credentials(force_refresh=True)
+                        refreshed = resolve_xai_http_credentials(
+                            force_refresh=True,
+                            api_key_hint=api_key,
+                        )
                         refreshed_key = str(refreshed.get("api_key") or "").strip()
                         if refreshed_key and refreshed_key != api_key:
                             api_key = refreshed_key

tests/agent/test_credential_pool.py

@@ -2970,6 +2970,78 @@ def _transient_failure(*_args, **_kwargs):
     assert tokens.get("refresh_token") == "old-refresh-token"
 
 
+def test_xai_oauth_concurrent_pool_instances_refresh_single_use_token_once(
+    tmp_path, monkeypatch
+):
+    import threading
+    import time
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
+    monkeypatch.delenv("XAI_API_KEY", raising=False)
+    monkeypatch.delenv("XAI_OAUTH_ACCESS_TOKEN", raising=False)
+
+    _write_auth_store(tmp_path, {
+        "version": 1,
+        "providers": {},
+        "credential_pool": {
+            "xai-oauth": [{
+                "id": "manual-xai",
+                "label": "manual-xai",
+                "auth_type": "oauth",
+                "priority": 0,
+                "source": "manual:xai_pkce",
+                "access_token": "old-access-token",
+                "refresh_token": "one-time-refresh-token",
+                "base_url": "https://api.x.ai/v1",
+            }],
+        },
+    })
+
+    from agent.credential_pool import load_pool
+    import hermes_cli.auth as auth_mod
+
+    pools = [load_pool("xai-oauth"), load_pool("xai-oauth")]
+    start = threading.Barrier(2)
+    refresh_calls: list[tuple[str, str]] = []
+
+    def _refresh(access_token, refresh_token, **_kwargs):
+        refresh_calls.append((access_token, refresh_token))
+        time.sleep(0.1)
+        return {
+            "access_token": "fresh-access-token",
+            "refresh_token": "fresh-refresh-token",
+            "last_refresh": "2026-07-12T00:00:00+00:00",
+        }
+
+    monkeypatch.setattr(auth_mod, "refresh_xai_oauth_pure", _refresh)
+    results = []
+    errors = []
+
+    def _worker(pool):
+        try:
+            start.wait()
+            results.append(pool.try_refresh_matching("old-access-token"))
+        except Exception as exc:
+            errors.append(exc)
+
+    threads = [threading.Thread(target=_worker, args=(pool,)) for pool in pools]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert not errors
+    assert refresh_calls == [("old-access-token", "one-time-refresh-token")]
+    assert sorted(entry.access_token for entry in results) == [
+        "fresh-access-token",
+        "fresh-access-token",
+    ]
+    persisted = json.loads((tmp_path / "hermes" / "auth.json").read_text())
+    stored = persisted["credential_pool"]["xai-oauth"][0]
+    assert stored["access_token"] == "fresh-access-token"
+    assert stored["refresh_token"] == "fresh-refresh-token"
+
+
 # ---------------------------------------------------------------------------
 # Codex OAuth terminal error quarantine
 # ---------------------------------------------------------------------------

tests/agent/test_credential_pool_routing.py

@@ -162,7 +162,7 @@ def _make_agent_with_pool(self, pool_entries=3):
         # mark_exhausted_and_rotate returns next entry until exhausted
         self._rotation_index = 0
 
-        def rotate(status_code=None, error_context=None):
+        def rotate(status_code=None, error_context=None, api_key_hint=None):
             self._rotation_index += 1
             if self._rotation_index < pool_entries:
                 return entries[self._rotation_index]
@@ -220,7 +220,11 @@ def test_402_immediate_rotation(self):
         )
         assert recovered is True
         assert has_retried is False
-        pool.mark_exhausted_and_rotate.assert_called_once_with(status_code=402, error_context=None)
+        pool.mark_exhausted_and_rotate.assert_called_once_with(
+            status_code=402,
+            error_context=None,
+            api_key_hint=None,
+        )
 
     def test_no_pool_returns_false(self):
         """No pool should return (False, unchanged)."""