fix(grok): route CLI traffic to api.x.ai on shared proxies

headroom/providers/grok/__init__.py

@@ -1,5 +1,17 @@
 """Grok CLI provider helpers."""
 
-from .runtime import DEFAULT_API_URL, PROXY_ENV_KEY, build_launch_env, proxy_base_url
+from .runtime import (
+    DEFAULT_API_URL,
+    PROXY_ENV_KEY,
+    build_launch_env,
+    is_grok_cli_request,
+    proxy_base_url,
+)
 
-__all__ = ["DEFAULT_API_URL", "PROXY_ENV_KEY", "build_launch_env", "proxy_base_url"]
+__all__ = [
+    "DEFAULT_API_URL",
+    "PROXY_ENV_KEY",
+    "build_launch_env",
+    "is_grok_cli_request",
+    "proxy_base_url",
+]

headroom/providers/grok/runtime.py

@@ -10,6 +10,43 @@
 DEFAULT_API_URL = "https://api.x.ai"
 PROXY_ENV_KEY = "GROK_MODELS_BASE_URL"
 
+# Official Grok CLI / Grok Build stamps this on inference requests (observed on
+# grok-shell 0.2.x). Used for per-request xAI routing when the shared proxy's
+# process-wide OPENAI target is still api.openai.com (Claude/Codex-started).
+_XAI_TOKEN_AUTH_HEADER = "x-xai-token-auth"
+_XAI_TOKEN_AUTH_VALUE = "xai-grok-cli"
+# UA product tokens emitted by Grok CLI 0.2.x. Matched against whitespace-split
+# tokens so unrelated clients ("litellm-grok/1.0") cannot collide.
+_GROK_UA_PREFIXES = ("grok-pager/", "grok-shell/")
+
+
+def _header_value(headers: Mapping[str, str], name: str) -> str | None:
+    """Case-insensitive header lookup for plain mappings and Starlette Headers."""
+    lowered = name.lower()
+    for key, value in headers.items():
+        if key.lower() == lowered:
+            return value
+    return None
+
+
+def is_grok_cli_request(headers: Mapping[str, str]) -> bool:
+    """Return True when inbound headers identify the official Grok CLI.
+
+    Grok cannot stamp ``x-headroom-base-url`` (no custom attribution headers),
+    so shared-proxy routing must recognize the CLI from wire signals instead.
+    Detection is intentionally narrow: only the official token-auth marker and
+    known Grok UA product tokens (prefix match on whitespace-split tokens) —
+    never model-id heuristics.
+    """
+    token_auth = _header_value(headers, _XAI_TOKEN_AUTH_HEADER)
+    if token_auth is not None and token_auth.strip().lower() == _XAI_TOKEN_AUTH_VALUE:
+        return True
+
+    user_agent = _header_value(headers, "user-agent")
+    if not user_agent:
+        return False
+    return any(token.startswith(_GROK_UA_PREFIXES) for token in user_agent.lower().split())
+
 
 def proxy_base_url(port: int) -> str:
     """Return the local proxy base URL used by Grok CLI integrations."""

headroom/providers/proxy_routes.py

@@ -38,6 +38,9 @@
 from headroom.providers.proxy_targets import (
     api_target as _api_target,
 )
+from headroom.providers.proxy_targets import (
+    openai_compatible_base_url as _openai_compatible_base_url,
+)
 from headroom.providers.proxy_targets import (
     select_passthrough_base_url as _select_passthrough_base_url,
 )
@@ -466,23 +469,35 @@ async def vertex_stream_raw_predict_no_version(
 
     @app.get("/v1/models")
     async def list_models(request: Request):
-        provider_name = proxy.provider_runtime.model_metadata_provider(dict(request.headers))
+        headers = dict(request.headers)
+        provider_name = proxy.provider_runtime.model_metadata_provider(headers)
+        base_url = (
+            _openai_compatible_base_url(proxy, headers)
+            if provider_name == "openai"
+            else _api_target(proxy, provider_name)
+        )
         return await handle_model_metadata_endpoint(
             proxy,
             request,
             endpoint=MODEL_METADATA_LIST_ENDPOINT,
-            provider_api_base_url=_api_target(proxy, provider_name),
+            provider_api_base_url=base_url,
             provider_name=provider_name,
         )
 
     @app.get("/v1/models/{model_id}")
     async def get_model(request: Request, model_id: str):
-        provider_name = proxy.provider_runtime.model_metadata_provider(dict(request.headers))
+        headers = dict(request.headers)
+        provider_name = proxy.provider_runtime.model_metadata_provider(headers)
+        base_url = (
+            _openai_compatible_base_url(proxy, headers)
+            if provider_name == "openai"
+            else _api_target(proxy, provider_name)
+        )
         return await handle_model_metadata_endpoint(
             proxy,
             request,
             endpoint=model_metadata_get_endpoint(model_id),
-            provider_api_base_url=_api_target(proxy, provider_name),
+            provider_api_base_url=base_url,
             provider_name=provider_name,
         )
 

headroom/providers/proxy_targets.py

@@ -7,6 +7,9 @@
 
 from headroom.providers.codex import resolve_codex_routing
 from headroom.providers.codex.endpoints import CHATGPT_BACKEND_API_URL
+from headroom.providers.codex.runtime import DEFAULT_API_URL as DEFAULT_OPENAI_API_URL
+from headroom.providers.grok.runtime import DEFAULT_API_URL as XAI_API_URL
+from headroom.providers.grok.runtime import is_grok_cli_request
 from headroom.providers.vertex import vertex_target_for_location as _vertex_target_for_location
 
 LEGACY_API_TARGET_ATTRS: dict[str, str] = {
@@ -29,6 +32,43 @@ def vertex_target_for_location(proxy: Any, location: str) -> str:
     return _vertex_target_for_location(api_target(proxy, "vertex"), location)
 
 
+def route_grok_to_xai(headers: Mapping[str, str], openai_target: str) -> bool:
+    """Return True when Grok CLI traffic should be redirected to ``api.x.ai``.
+
+    Grok CLI cannot set ``x-headroom-base-url``, so a shared proxy started for
+    Claude/Codex has to recognize it from wire signals or it forwards xAI
+    session tokens to ``api.openai.com``.
+
+    Only applies while the OpenAI target is still the default. An operator who
+    pointed the proxy at a gateway (LiteLLM, Azure, self-hosted vLLM) chose it
+    for every OpenAI-compatible client; a client User-Agent must not silently
+    bypass that.
+
+    This gate is URL policy only. It does not keep operator-configured
+    ``OPENAI_TARGET_API_HEADERS`` away from xAI — those extras are configured
+    independently of the target URL, so a default-URL proxy can still redirect
+    here. The direct OpenAI HTTP handlers enforce credential isolation
+    separately by suppressing configured extras when their OpenAI-compatible
+    upstream candidate is the xAI host. Configured backend transports retain
+    their existing header policy.
+    """
+    if not is_grok_cli_request(headers):
+        return False
+    return openai_target.rstrip("/") == DEFAULT_OPENAI_API_URL
+
+
+def openai_compatible_base_url(proxy: Any, headers: Mapping[str, str]) -> str:
+    """Resolve upstream for OpenAI-compatible metadata/passthrough traffic.
+
+    Routes official Grok CLI to ``api.x.ai`` so ``GET /v1/models`` and catch-all
+    passthrough succeed on a shared proxy whose OpenAI target is the default.
+    """
+    target = api_target(proxy, "openai")
+    if route_grok_to_xai(headers, target):
+        return XAI_API_URL
+    return target
+
+
 def select_passthrough_base_url(proxy: Any, headers: Mapping[str, str]) -> str:
     """Resolve the upstream base URL for catch-all proxy passthrough requests."""
     routing = resolve_codex_routing(headers)
@@ -41,4 +81,6 @@ def select_passthrough_base_url(proxy: Any, headers: Mapping[str, str]) -> str:
         if azure_base:
             return azure_base.rstrip("/")
     provider_name = proxy.provider_runtime.model_metadata_provider(headers)
+    if provider_name == "openai":
+        return openai_compatible_base_url(proxy, headers)
     return api_target(proxy, provider_name)

headroom/proxy/handlers/openai.py

@@ -65,6 +65,8 @@
     resolve_codex_routing_headers as _resolve_codex_routing_headers,
 )
 from headroom.providers.copilot import model_prefers_responses_api
+from headroom.providers.grok.runtime import DEFAULT_API_URL as XAI_API_URL
+from headroom.providers.proxy_targets import route_grok_to_xai
 from headroom.proxy.auth_mode import (
     classify_auth_mode,
     classify_client,
@@ -394,6 +396,26 @@ def _append_request_query(url: str, query: str) -> str:
     return f"{url}{separator}{query}"
 
 
+def _xai_hostname(url: str) -> str | None:
+    """Return ``url``'s hostname with any fully-qualified trailing dot removed.
+
+    ``https://api.x.ai.`` resolves to the same host as ``https://api.x.ai`` but
+    ``urlparse`` reports a distinct hostname, so a client-supplied
+    ``x-headroom-base-url`` could otherwise slip past the comparison below.
+    """
+    hostname = urlparse(url).hostname
+    return hostname.rstrip(".") if hostname else hostname
+
+
+def _is_xai_upstream(upstream_base_url: str) -> bool:
+    """Return whether the selected upstream is the official xAI API host.
+
+    Compares the parsed hostname only: a path, scheme or port variation of
+    ``api.x.ai`` is still xAI and must not receive OpenAI-side credentials.
+    """
+    return _xai_hostname(upstream_base_url) == _xai_hostname(XAI_API_URL)
+
+
 def _normalize_origin(origin: str) -> str | None:
     parsed = urlparse(origin.strip())
     if not parsed.scheme or not parsed.hostname:
@@ -1703,10 +1725,36 @@ def _resolve_openai_upstream(self, request: Request) -> str:
         Honors the ``x-headroom-base-url`` request header so OpenAI-compatible
         gateways (LiteLLM, CPA, self-hosted vLLM, Azure OpenAI) route through
         the dedicated ``/v1/chat/completions`` and ``/v1/responses`` handlers,
-        not just the generic passthrough route that already honors it. Falls
-        back to the configured ``OPENAI_API_URL`` (``OPENAI_TARGET_API_URL``).
+        not just the generic passthrough route that already honors it.
+
+        When the header is absent, official Grok CLI requests (identified by
+        ``x-xai-token-auth`` / Grok UA tokens) route to ``api.x.ai`` so a
+        shared proxy started for Claude/Codex does not forward Grok session
+        tokens to ``api.openai.com`` — but only while ``OPENAI_API_URL`` is
+        still the default, so a configured gateway is never bypassed.
+        Otherwise falls back to the configured ``OPENAI_API_URL``
+        (``OPENAI_TARGET_API_URL``).
+        """
+        custom = _resolve_openai_upstream_base(request.headers)
+        if custom is not None:
+            return custom
+        if route_grok_to_xai(request.headers, self.OPENAI_API_URL):
+            return XAI_API_URL
+        return self.OPENAI_API_URL
+
+    def _openai_extra_headers_for_upstream(self, upstream_base_url: str) -> dict[str, str] | None:
+        """Return configured OpenAI extras for a direct non-xAI upstream.
+
+        ``openai_extra_headers`` is operator-owned and scoped to the OpenAI
+        target (an API key for a gateway, a tenant header, ...). ``api.x.ai`` is
+        reached with the *client's* own xAI credential, so those extras must
+        never travel there — ``merge_extra_headers`` overrides same-named keys,
+        so a configured ``Authorization`` would both leak the operator's OpenAI
+        credential and clobber the client's ``Bearer xai-...``.
         """
-        return _resolve_openai_upstream_base(request.headers) or self.OPENAI_API_URL
+        if _is_xai_upstream(upstream_base_url):
+            return None
+        return self.config.openai_extra_headers
 
     @staticmethod
     def _strict_previous_turn_frozen_count(
@@ -3119,25 +3167,33 @@ async def handle_openai_chat(
 
         _pre_strip_count_chat = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.openai_extra_headers)
+        # Configured backends own their destination and authentication, so they
+        # retain the existing extra-header policy. The direct path selects
+        # extras from the resolved OpenAI-compatible upstream.
+        headers = merge_extra_headers(
+            headers,
+            self.config.openai_extra_headers
+            if self.anthropic_backend is not None
+            else self._openai_extra_headers_for_upstream(upstream_base_url),
+        )
         log_outbound_headers(
             forwarder="openai_chat_completions",
             stripped_count=_pre_strip_count_chat,
             request_id=request_id,
         )
-        upstream_base_url = _resolve_openai_upstream_base(request.headers)
+        custom_upstream_base_url = _resolve_openai_upstream_base(request.headers)
         handler_path = (
             _resolve_openai_handler_path(
                 request.headers,
                 handler_path=_OPENAI_CHAT_COMPLETIONS_PATH,
             )
-            if upstream_base_url is not None
+            if custom_upstream_base_url is not None
             else "/v1/chat/completions"
         )
         _, custom_chat_provider = _custom_base_passthrough_telemetry(
             request.method,
             handler_path,
-            upstream_base_url or "",
+            custom_upstream_base_url or "",
         )
         openai_chat_outcome_provider = custom_chat_provider or "openai"
 
@@ -4441,9 +4497,13 @@ async def api_call_fn(
                     },
                 )
 
-        # Direct OpenAI API (no backend configured)
+        # Direct OpenAI API (no backend configured). Reuse the upstream resolved
+        # once at request entry (custom base → Grok CLI → process default): the
+        # local ``custom_upstream_base_url`` above is custom-header only, and the
+        # same value already decided which extra headers were merged, so routing
+        # and header policy cannot drift apart.
         url = build_copilot_upstream_url(
-            upstream_base_url or self.OPENAI_API_URL,
+            upstream_base_url,
             handler_path,
         )
         url = _append_request_query(url, request.url.query)
@@ -5078,7 +5138,14 @@ async def handle_openai_responses(
 
         _pre_strip_count_resp = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
         headers = _strip_internal_headers(headers)
-        headers = merge_extra_headers(headers, self.config.openai_extra_headers)
+        # Resolve the OpenAI-compatible candidate before merging operator
+        # extras. Mixed ChatGPT-auth + Grok-signal requests conservatively
+        # withhold extras based on the xAI candidate even though the higher-
+        # priority routing branch below still sends them to chatgpt.com.
+        openai_upstream_base_url = self._resolve_openai_upstream(request)
+        headers = merge_extra_headers(
+            headers, self._openai_extra_headers_for_upstream(openai_upstream_base_url)
+        )
         # Mirror the WS handler: never forward Codex's client-only lite header
         # upstream. OpenAI rejects newer Codex models when it leaks, and the HTTP
         # POST path (unlike the WS path) otherwise forwards request headers verbatim.
@@ -5378,14 +5445,17 @@ async def handle_openai_responses(
         if is_chatgpt_auth:
             url = codex_responses_http_url()
         else:
-            upstream_base_url = _resolve_openai_upstream_base(request.headers)
+            custom_upstream_base_url = _resolve_openai_upstream_base(request.headers)
             handler_path = (
                 _resolve_openai_handler_path(request.headers, handler_path=_OPENAI_RESPONSES_PATH)
-                if upstream_base_url is not None
+                if custom_upstream_base_url is not None
                 else "/v1/responses"
             )
+            # Reuse the OpenAI-compatible candidate resolved at request entry.
+            # In this non-ChatGPT branch it is also the actual upstream, keeping
+            # direct routing and header selection aligned.
             url = build_copilot_upstream_url(
-                upstream_base_url or self.OPENAI_API_URL,
+                openai_upstream_base_url,
                 handler_path,
             )
             url = _append_request_query(url, request.url.query)