fix(auxiliary): detect xAI OAuth 403 bad-credentials as auth error

agent/auxiliary_client.py

@@ -2347,7 +2347,16 @@ def _is_auth_error(exc: Exception) -> bool:
     if status == 401:
         return True
     err_lower = str(exc).lower()
-    return "error code: 401" in err_lower or "authenticationerror" in type(exc).__name__.lower()
+    if "error code: 401" in err_lower or "authenticationerror" in type(exc).__name__.lower():
+        return True
+    # xAI returns HTTP 403 with "unauthenticated:bad-credentials" when an OAuth2
+    # access token has expired or is invalid — semantically a 401 auth failure,
+    # even though the status code is 403 (PermissionDenied).
+    if status == 403 and "bad-credentials" in err_lower:
+        return True
+    if "unauthenticated" in err_lower and "bad-credentials" in err_lower:
+        return True
+    return False
 
 
 def _is_unsupported_parameter_error(exc: Exception, param: str) -> bool:
@@ -2496,6 +2505,8 @@ def _recoverable_pool_provider(resolved_provider: str, client: Any) -> Optional[
         return "copilot"
     if base_url_host_matches(base, "api.kimi.com"):
         return "kimi-coding"
+    if base_url_host_matches(base, "api.x.ai"):
+        return "xai-oauth"
     return None
 
 
@@ -2692,6 +2703,24 @@ def _refresh_provider_credentials(provider: str) -> bool:
                 return False
             _evict_cached_clients(normalized)
             return True
+        if normalized == "xai-oauth":
+            # Preference: pool-level refresh (uses refresh_token from pool entry),
+            # then fall back to singleton auth-store resolver.
+            pool = load_pool(normalized)
+            if pool and pool.has_credentials():
+                # Ensure a current entry is selected before trying to refresh.
+                pool.select()
+                refreshed = pool.try_refresh_current()
+                if refreshed is not None and str(getattr(refreshed, "runtime_api_key", "") or "").strip():
+                    _evict_cached_clients(normalized)
+                    return True
+            from hermes_cli.auth import resolve_xai_oauth_runtime_credentials
+
+            creds = resolve_xai_oauth_runtime_credentials(force_refresh=True)
+            if not str(creds.get("api_key", "") or "").strip():
+                return False
+            _evict_cached_clients(normalized)
+            return True
     except Exception as exc:
         logger.debug("Auxiliary provider credential refresh failed for %s: %s", normalized, exc)
         return False

tests/agent/test_auxiliary_client_xai_oauth_recovery.py

@@ -0,0 +1,153 @@
+"""Tests for xAI OAuth 403 error recovery in auxiliary_client.
+
+xAI returns HTTP 403 (not 401) with "unauthenticated:bad-credentials" when
+an OAuth2 access token has expired.  These tests verify the three fixes:
+
+1. _is_auth_error detects xAI 403 as an auth failure
+2. _recoverable_pool_provider maps api.x.ai to xai-oauth
+3. _refresh_provider_credentials includes xai-oauth refresh logic
+"""
+
+import pytest
+
+
+# ── _is_auth_error ──────────────────────────────────────────────────────────
+
+def _import_is_auth_error():
+    from agent.auxiliary_client import _is_auth_error
+    return _is_auth_error
+
+
+class TestIsAuthErrorXaiOauth403:
+    """Verify _is_auth_error correctly identifies xAI's 403 bad-credentials."""
+
+    @pytest.fixture(autouse=True)
+    def _import(self):
+        self.is_auth_error = _import_is_auth_error()
+
+    def test_xai_403_bad_credentials_is_auth_error(self):
+        """The exact error xAI returns for expired OAuth tokens."""
+        exc = Exception(
+            "Error code: 403 - {'code': 'The caller does not have permission "
+            "to execute the specified operation', 'error': 'The OAuth2 access "
+            "token could not be validated. [WKE=unauthenticated:bad-credentials]'}"
+        )
+        exc.status_code = 403  # openai.PermissionDenied sets this
+        assert self.is_auth_error(exc) is True
+
+    def test_xai_403_bad_credentials_without_status_code(self):
+        """Fallback match when status_code attribute is missing."""
+        exc = Exception(
+            "Error code: 403 - unauthenticated:bad-credentials"
+        )
+        # No status_code attribute — should still match via string pattern
+        assert self.is_auth_error(exc) is True
+
+    def test_generic_403_is_not_auth_error(self):
+        """A generic 403 (e.g. rate limit, forbidden) should NOT be treated as auth."""
+        exc = Exception("Error code: 403 - rate limit exceeded")
+        exc.status_code = 403
+        assert self.is_auth_error(exc) is False
+
+    def test_401_status_code_is_auth_error(self):
+        """Existing 401 detection still works."""
+        exc = Exception("Unauthorized")
+        exc.status_code = 401
+        assert self.is_auth_error(exc) is True
+
+    def test_401_string_is_auth_error(self):
+        """Existing string-based 401 detection still works."""
+        exc = Exception("Error code: 401 - Unauthorized")
+        assert self.is_auth_error(exc) is True
+
+    def test_authentication_error_class_is_auth_error(self):
+        """Existing AuthenticationError class detection still works."""
+        exc_type = type("AuthenticationError", (Exception,), {})
+        exc = exc_type("auth failure")
+        assert self.is_auth_error(exc) is True
+
+    def test_permission_denied_without_bad_credentials_is_not_auth_error(self):
+        """403 PermissionDenied without bad-credentials should not be auth."""
+        exc = Exception("Error code: 403 - Permission denied")
+        exc.status_code = 403
+        assert self.is_auth_error(exc) is False
+
+    def test_500_is_not_auth_error(self):
+        """Server errors are not auth errors."""
+        exc = Exception("Error code: 500 - Internal server error")
+        exc.status_code = 500
+        assert self.is_auth_error(exc) is False
+
+    def test_unauthenticated_without_bad_credentials_is_not_auth_error(self):
+        """'unauthenticated' alone (without 'bad-credentials') should not match."""
+        exc = Exception("unauthenticated request")
+        assert self.is_auth_error(exc) is False
+
+
+# ── _recoverable_pool_provider ──────────────────────────────────────────────
+
+def _import_recoverable_pool_provider():
+    from agent.auxiliary_client import _recoverable_pool_provider
+    return _recoverable_pool_provider
+
+
+class TestRecoverablePoolProviderXaiOAuth:
+    """Verify _recoverable_pool_provider maps api.x.ai to xai-oauth."""
+
+    @pytest.fixture(autouse=True)
+    def _import(self):
+        self.recover = _import_recoverable_pool_provider()
+
+    def test_explicit_xai_oauth_provider(self):
+        """Explicit provider name passes through."""
+        result = self.recover("xai-oauth", None)
+        assert result == "xai-oauth"
+
+    def test_api_x_ai_host_match(self):
+        """api.x.ai base URL maps to xai-oauth pool."""
+        class MockClient:
+            base_url = "https://api.x.ai/v1/"
+
+        result = self.recover("auto", MockClient())
+        assert result == "xai-oauth"
+
+    def test_auto_with_unknown_host_returns_none(self):
+        """auto provider with unknown host returns None."""
+        class MockClient:
+            base_url = "https://unknown.example.com/v1/"
+
+        result = self.recover("auto", MockClient())
+        assert result is None
+
+
+# ── _refresh_provider_credentials (structure check) ─────────────────────────
+
+def _import_refresh_provider_credentials():
+    from agent.auxiliary_client import _refresh_provider_credentials
+    return _refresh_provider_credentials
+
+
+class TestRefreshProviderCredentialsXaiOAuth:
+    """Verify _refresh_provider_credentials has xai-oauth branch.
+
+    Full integration testing requires live OAuth tokens, so we verify
+    the branch exists and handles the no-credential case gracefully.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _import(self):
+        self.refresh = _import_refresh_provider_credentials()
+
+    def test_xai_oauth_no_pool_returns_false(self):
+        """When no xai-oauth pool exists, refresh returns False gracefully."""
+        # This tests that the branch exists and doesn't crash.
+        # It may return True if the singleton resolver finds tokens,
+        # or False if neither pool nor singleton has credentials.
+        # Either way, it should not raise an exception.
+        result = self.refresh("xai-oauth")
+        assert isinstance(result, bool)
+
+    def test_unknown_provider_returns_false(self):
+        """Unknown providers fall through to return False."""
+        result = self.refresh("unknown-provider-xyz")
+        assert result is False