fix(stt): retry xAI OAuth after auth rejection

tests/tools/test_transcription_tools.py

@@ -11,7 +11,7 @@ import struct
 import subprocess
 import types
 import wave
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
 import pytest
 
@@ -1570,6 +1570,59 @@ class TestTranscribeXAI:
         assert "HTTP 400" in result["error"]
         assert "Invalid audio format" in result["error"]
 
+    def test_retries_403_with_refreshed_oauth_credentials(
+        self, sample_ogg, mock_xai_http_module
+    ):
+        mock_xai_http_module.resolve_xai_http_credentials.side_effect = [
+            {
+                "api_key": "stale-oauth-token",
+                "base_url": "https://api.x.ai/v1",
+                "provider": "xai-oauth",
+            },
+            {
+                "api_key": "fresh-oauth-token",
+                "base_url": "https://api.x.ai/v1",
+                "provider": "xai-oauth",
+            },
+        ]
+
+        rejected = MagicMock()
+        rejected.status_code = 403
+        rejected.json.return_value = {
+            "error": {"message": "OAuth2 access token could not be validated"}
+        }
+        accepted = MagicMock()
+        accepted.status_code = 200
+        accepted.json.return_value = {
+            "text": "fleet speech transcription proof",
+            "language": "en",
+            "duration": 2.1,
+        }
+
+        stt_config = {"provider": "xai"}
+        with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
+             patch("tools.transcription_tools._get_provider", return_value="xai"), \
+             patch("requests.post", side_effect=[rejected, accepted]) as mock_post:
+            from tools.transcription_tools import transcribe_audio
+            result = transcribe_audio(sample_ogg)
+
+        assert result == {
+            "success": True,
+            "transcript": "fleet speech transcription proof",
+            "provider": "xai",
+        }
+        assert mock_post.call_count == 2
+        assert mock_post.call_args_list[0].kwargs["headers"]["Authorization"] == (
+            "Bearer stale-oauth-token"
+        )
+        assert mock_post.call_args_list[1].kwargs["headers"]["Authorization"] == (
+            "Bearer fresh-oauth-token"
+        )
+        assert mock_xai_http_module.resolve_xai_http_credentials.call_args_list == [
+            call(),
+            call(force_refresh=True, api_key_hint="stale-oauth-token"),
+        ]
+
     def test_empty_transcript_returns_failure(self, monkeypatch, sample_ogg, mock_xai_http_module):
         monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
 

tools/transcription_tools.py

@@ -1819,12 +1819,16 @@ def _transcribe_xai(file_path: str, model_name: str) -> Dict[str, Any]:
 
     stt_config = _load_stt_config()
     xai_config = stt_config.get("xai") or {}
-    base_url = str(
-        xai_config.get("base_url")
-        or get_env_value("XAI_STT_BASE_URL")
-        or creds.get("base_url")
-        or XAI_STT_BASE_URL
-    ).strip().rstrip("/")
+
+    def _resolve_base_url(resolved_creds: Dict[str, str]) -> str:
+        return str(
+            xai_config.get("base_url")
+            or get_env_value("XAI_STT_BASE_URL")
+            or resolved_creds.get("base_url")
+            or XAI_STT_BASE_URL
+        ).strip().rstrip("/")
+
+    base_url = _resolve_base_url(creds)
     language = _resolve_stt_language("xai", stt_config) or ""
     # .get("format", True) already defaults to True when the key is absent;
     # is_truthy_value only normalizes truthy/falsy strings from config.
@@ -1843,19 +1847,48 @@ def _transcribe_xai(file_path: str, model_name: str) -> Dict[str, Any]:
         if use_diarize:
             data["diarize"] = "true"
 
-        with open(file_path, "rb") as audio_file:
-            response = requests.post(
-                f"{base_url}/stt",
-                headers={
-                    "Authorization": f"Bearer {api_key}",
-                    "User-Agent": hermes_xai_user_agent(),
-                },
-                files={
-                    "file": (Path(file_path).name, audio_file),
-                },
-                data=data,
-                timeout=120,
+        def _post_transcription(bearer: str, endpoint_base_url: str):
+            with open(file_path, "rb") as audio_file:
+                return requests.post(
+                    f"{endpoint_base_url}/stt",
+                    headers={
+                        "Authorization": f"Bearer {bearer}",
+                        "User-Agent": hermes_xai_user_agent(),
+                    },
+                    files={
+                        "file": (Path(file_path).name, audio_file),
+                    },
+                    data=data,
+                    timeout=120,
+                )
+
+        response = _post_transcription(api_key, base_url)
+
+        if (
+            response.status_code in {401, 403}
+            and creds.get("provider") == "xai-oauth"
+        ):
+            logger.info(
+                "xAI STT got HTTP %d; refreshing OAuth credentials and retrying once",
+                response.status_code,
             )
+            try:
+                refreshed_creds = resolve_xai_http_credentials(
+                    force_refresh=True,
+                    api_key_hint=api_key,
+                )
+                refreshed_key = str(refreshed_creds.get("api_key") or "").strip()
+                if refreshed_key and refreshed_key != api_key:
+                    response = _post_transcription(
+                        refreshed_key,
+                        _resolve_base_url(refreshed_creds),
+                    )
+            except Exception as refresh_exc:
+                logger.warning(
+                    "xAI STT OAuth refresh after HTTP %d failed: %s",
+                    response.status_code,
+                    refresh_exc,
+                )
 
         if response.status_code != 200:
             detail = ""