Skip to content
Merged
72 changes: 72 additions & 0 deletions agent/auxiliary_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1254,6 +1254,30 @@ def _resolve_nous_runtime_api(*, force_refresh: bool = False) -> Optional[tuple[
return api_key, base_url


def _resolve_xai_oauth_for_aux() -> Optional[Tuple[str, str]]:
"""Resolve a fresh xAI OAuth (api_key, base_url) for auxiliary clients.

Routes through ``hermes_cli.auth``'s runtime resolver so the auto-refresh
path is shared with the main agent, instead of relying on whatever raw
tokens happen to be sitting in auth.json or the credential pool. Returns
``None`` if the user is not authenticated with xAI Grok OAuth (so
``_resolve_auto`` Step 1 falls through to the next provider in the chain).
"""
try:
from hermes_cli.auth import resolve_xai_oauth_runtime_credentials

creds = resolve_xai_oauth_runtime_credentials()
except Exception as exc:
logger.debug("Auxiliary xAI OAuth runtime credential resolution failed: %s", exc)
return None

api_key = str(creds.get("api_key") or "").strip()
base_url = str(creds.get("base_url") or "").strip().rstrip("/")
if not api_key or not base_url:
return None
return api_key, base_url


def _read_codex_access_token() -> Optional[str]:
"""Read a valid, non-expired Codex OAuth access token from Hermes auth store.

Expand Down Expand Up @@ -1744,6 +1768,32 @@ def _try_custom_endpoint() -> Tuple[Optional[Any], Optional[str]]:
return _fallback_client, model


def _build_xai_oauth_aux_client(model: str) -> Tuple[Optional[Any], Optional[str]]:
"""Build a CodexAuxiliaryClient for an xAI Grok OAuth-authenticated session.

xAI's ``/v1/responses`` endpoint speaks the OpenAI Responses API, so we
wrap a plain ``OpenAI`` client in ``CodexAuxiliaryClient`` to translate
``chat.completions.create()`` calls into ``responses.stream()`` requests.

The caller must pass an explicit model — pinning a default for Grok
would silently rot when xAI's allowlist drifts. Returns ``(None, None)``
when the user has not authenticated with xAI Grok OAuth.
"""
if not model:
logger.warning(
"Auxiliary client: xai-oauth requested without a model; "
"pass model explicitly (auxiliary.<task>.model in config.yaml)."
)
return None, None
resolved = _resolve_xai_oauth_for_aux()
if resolved is None:
return None, None
api_key, base_url = resolved
logger.debug("Auxiliary client: xAI OAuth (%s via Responses API)", model)
real_client = OpenAI(api_key=api_key, base_url=base_url)
return CodexAuxiliaryClient(real_client, model), model


def _build_codex_client(model: str) -> Tuple[Optional[Any], Optional[str]]:
"""Build a CodexAuxiliaryClient for an explicitly-requested model.

Expand Down Expand Up @@ -2851,6 +2901,26 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = "",
return (_to_async_client(client, final_model, is_vision=is_vision) if async_mode
else (client, final_model))

# ── xAI Grok OAuth (loopback PKCE → Responses API) ───────────────
# Without this branch, an xai-oauth main provider falls through to the
# generic ``oauth_external`` arm below and returns ``(None, None)``,
# silently re-routing every auxiliary task (compression, web extract,
# session search, curator, etc.) to whatever Step-2 fallback the user
# has configured. Users on xAI Grok OAuth would then see surprise
# OpenRouter / Nous bills for side tasks they thought were running on
# their xAI subscription.
if provider == "xai-oauth":
client, default = _build_xai_oauth_aux_client(model)
if client is None:
logger.warning(
"resolve_provider_client: xai-oauth requested but no xAI "
"OAuth token found (run: hermes model -> xAI Grok OAuth — SuperGrok Subscription)"
)
return None, None
final_model = _normalize_resolved_model(model or default, provider)
return (_to_async_client(client, final_model, is_vision=is_vision) if async_mode
else (client, final_model))

# ── Custom endpoint (OPENAI_BASE_URL + OPENAI_API_KEY) ───────────
if provider == "custom":
if explicit_base_url:
Expand Down Expand Up @@ -3201,6 +3271,8 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = "",
return resolve_provider_client("nous", model, async_mode)
if provider == "openai-codex":
return resolve_provider_client("openai-codex", model, async_mode)
if provider == "xai-oauth":
return resolve_provider_client("xai-oauth", model, async_mode)
# Other OAuth providers not directly supported
logger.warning("resolve_provider_client: OAuth provider %s not "
"directly supported, try 'auto'", provider)
Expand Down
15 changes: 14 additions & 1 deletion agent/codex_responses_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -726,7 +726,7 @@ def _preflight_codex_api_kwargs(
"model", "instructions", "input", "tools", "store",
"reasoning", "include", "max_output_tokens", "temperature",
"tool_choice", "parallel_tool_calls", "prompt_cache_key", "service_tier",
"extra_headers",
"extra_headers", "extra_body",
}
normalized: Dict[str, Any] = {
"model": model,
Expand Down Expand Up @@ -776,6 +776,19 @@ def _preflight_codex_api_kwargs(
if normalized_headers:
normalized["extra_headers"] = normalized_headers

extra_body = api_kwargs.get("extra_body")
if extra_body is not None:
if not isinstance(extra_body, dict):
raise ValueError("Codex Responses request 'extra_body' must be an object.")
# Pass extra_body through verbatim — used by xAI Responses to
# carry `prompt_cache_key` as a body-level field (the documented
# cache-routing surface on /v1/responses). The openai SDK
# serializes extra_body into the JSON body without per-field
# type checks, so it survives Responses.stream() kwarg-signature
# changes that would otherwise raise TypeError before the wire.
if extra_body:
normalized["extra_body"] = dict(extra_body)

if allow_stream:
stream = api_kwargs.get("stream")
if stream is not None and stream is not True:
Expand Down
184 changes: 180 additions & 4 deletions agent/credential_pool.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
_resolve_zai_base_url,
_save_auth_store,
_save_provider_state,
_store_provider_state,
read_credential_pool,
write_credential_pool,
)
Expand Down Expand Up @@ -539,6 +540,64 @@ def _sync_codex_entry_from_auth_store(self, entry: PooledCredential) -> PooledCr
logger.debug("Failed to sync Codex entry from auth.json: %s", exc)
return entry

def _sync_xai_oauth_entry_from_auth_store(self, entry: PooledCredential) -> PooledCredential:
"""Sync an xAI OAuth pool entry from auth.json if tokens differ.

xAI OAuth refresh tokens are single-use. When another Hermes process
(or another profile sharing the same auth.json) refreshes the token,
it writes the new pair to ``providers["xai-oauth"]["tokens"]`` under
``_auth_store_lock``. Without this resync, our in-memory pool entry
keeps the consumed refresh_token and the next ``_refresh_entry`` call
would replay it and get a ``refresh_token_reused``-style 4xx.

Only applies to entries seeded from the singleton (``loopback_pkce``);
manually added entries (``manual:xai_pkce``) are independent
credentials with their own refresh-token lifecycle.
"""
if self.provider != "xai-oauth" or entry.source != "loopback_pkce":
return entry
try:
with _auth_store_lock():
auth_store = _load_auth_store()
state = _load_provider_state(auth_store, "xai-oauth")
if not isinstance(state, dict):
return entry
tokens = state.get("tokens")
if not isinstance(tokens, dict):
return entry
store_access = tokens.get("access_token", "")
store_refresh = tokens.get("refresh_token", "")
entry_access = entry.access_token or ""
entry_refresh = entry.refresh_token or ""
if store_access and (
store_access != entry_access
or (store_refresh and store_refresh != entry_refresh)
):
logger.debug(
"Pool entry %s: syncing xAI OAuth tokens from auth.json "
"(refreshed by another process)",
entry.id,
)
field_updates: Dict[str, Any] = {
"access_token": store_access,
"refresh_token": store_refresh or entry.refresh_token,
"last_status": None,
"last_status_at": None,
"last_error_code": None,
"last_error_reason": None,
"last_error_message": None,
"last_error_reset_at": None,
}
if state.get("last_refresh"):
field_updates["last_refresh"] = state["last_refresh"]
updated = replace(entry, **field_updates)
self._replace_entry(entry, updated)
self._persist()
return updated
except Exception as exc:
logger.debug("Failed to sync xAI OAuth entry from auth.json: %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.

Expand Down Expand Up @@ -604,9 +663,22 @@ def _sync_device_code_entry_to_auth_store(self, entry: PooledCredential) -> None
re-seeding a consumed single-use refresh token.

Applies to any OAuth provider whose singleton lives in auth.json
(currently Nous and OpenAI Codex).
(currently Nous, OpenAI Codex, and xAI Grok OAuth).

``set_active=False`` on every write: a pool sync-back is a
token-rotation side effect, not the user choosing a provider.
Using ``_save_provider_state`` (which sets ``active_provider``)
here would mean every Nous/Codex/xAI refresh in a multi-provider
setup silently flips the ``active_provider`` flag — the next
``hermes`` invocation that defaults to the active provider
(e.g. setup wizard, ``hermes auth status``) would land on
whatever provider happened to refresh last, not whatever the
user actually chose.
"""
if entry.source != "device_code":
# Only sync entries that were seeded *from* a singleton. Manually
# added pool entries (source="manual:*") are independent credentials
# and must not write back to the singleton.
if entry.source not in {"device_code", "loopback_pkce"}:
return
try:
with _auth_store_lock():
Expand All @@ -632,7 +704,7 @@ def _sync_device_code_entry_to_auth_store(self, entry: PooledCredential) -> None
state[extra_key] = val
if entry.inference_base_url:
state["inference_base_url"] = entry.inference_base_url
_save_provider_state(auth_store, "nous", state)
_store_provider_state(auth_store, "nous", state, set_active=False)

elif self.provider == "openai-codex":
state = _load_provider_state(auth_store, "openai-codex")
Expand All @@ -646,7 +718,21 @@ def _sync_device_code_entry_to_auth_store(self, entry: PooledCredential) -> None
tokens["refresh_token"] = entry.refresh_token
if entry.last_refresh:
state["last_refresh"] = entry.last_refresh
_save_provider_state(auth_store, "openai-codex", state)
_store_provider_state(auth_store, "openai-codex", state, set_active=False)

elif self.provider == "xai-oauth":
state = _load_provider_state(auth_store, "xai-oauth")
if not isinstance(state, dict):
return
tokens = state.get("tokens")
if not isinstance(tokens, dict):
return
tokens["access_token"] = entry.access_token
if entry.refresh_token:
tokens["refresh_token"] = entry.refresh_token
if entry.last_refresh:
state["last_refresh"] = entry.last_refresh
_store_provider_state(auth_store, "xai-oauth", state, set_active=False)

else:
return
Expand Down Expand Up @@ -699,6 +785,25 @@ def _refresh_entry(self, entry: PooledCredential, *, force: bool) -> Optional[Po
refresh_token=refreshed["refresh_token"],
last_refresh=refreshed.get("last_refresh"),
)
elif self.provider == "xai-oauth":
# Adopt fresher tokens from auth.json before spending the
# refresh_token — single-use tokens consumed by another
# process (or another profile sharing the singleton) would
# otherwise trigger ``refresh_token_reused`` on the next
# POST. Only meaningful for singleton-seeded entries.
synced = self._sync_xai_oauth_entry_from_auth_store(entry)
if synced is not entry:
entry = synced
refreshed = auth_mod.refresh_xai_oauth_pure(
entry.access_token,
entry.refresh_token,
)
updated = replace(
entry,
access_token=refreshed["access_token"],
refresh_token=refreshed["refresh_token"],
last_refresh=refreshed.get("last_refresh"),
)
elif self.provider == "nous":
synced = self._sync_nous_entry_from_auth_store(entry)
if synced is not entry:
Expand Down Expand Up @@ -777,6 +882,30 @@ def _refresh_entry(self, entry: PooledCredential, *, force: bool) -> Optional[Po
# Credentials file had a valid (non-expired) token — use it directly
logger.debug("Credentials file has valid token, using without refresh")
return synced
# For xai-oauth: same race as nous — another process may have
# consumed the refresh token between our proactive sync and the
# HTTP call. Re-check auth.json and adopt the fresh tokens if
# they have rotated since. Only meaningful for singleton-seeded
# (loopback_pkce) entries; manual entries don't share state with
# the singleton.
if self.provider == "xai-oauth":
synced = self._sync_xai_oauth_entry_from_auth_store(entry)
if synced.refresh_token != entry.refresh_token:
logger.debug(
"xAI OAuth refresh failed but auth.json has newer tokens — adopting"
)
updated = replace(
synced,
last_status=STATUS_OK,
last_status_at=None,
last_error_code=None,
last_error_reason=None,
last_error_message=None,
last_error_reset_at=None,
)
self._replace_entry(synced, updated)
self._persist()
return updated
# For nous: another process may have consumed the refresh token
# between our proactive sync and the HTTP call. Re-sync from
# auth.json and adopt the fresh tokens if available.
Expand Down Expand Up @@ -829,6 +958,11 @@ def _entry_needs_refresh(self, entry: PooledCredential) -> bool:
entry.access_token,
CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
)
if self.provider == "xai-oauth":
return auth_mod._xai_access_token_is_expiring(
entry.access_token,
auth_mod.XAI_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
)
if self.provider == "nous":
# Nous refresh/mint can require network access and should happen when
# runtime credentials are actually resolved, not merely when the pool
Expand Down Expand Up @@ -883,6 +1017,17 @@ def _available_entries(self, *, clear_expired: bool = False, refresh: bool = Fal
if synced is not entry:
entry = synced
cleared_any = True
# For xai-oauth singleton-seeded entries, identical pattern:
# an entry frozen as exhausted may simply be holding stale
# tokens that another process (or a fresh `hermes model` ->
# xAI Grok OAuth login) has since rotated in auth.json.
if (self.provider == "xai-oauth"
and entry.source == "loopback_pkce"
and entry.last_status == STATUS_EXHAUSTED):
synced = self._sync_xai_oauth_entry_from_auth_store(entry)
if synced is not entry:
entry = synced
cleared_any = True
if entry.last_status == STATUS_EXHAUSTED:
exhausted_until = _exhausted_until(entry)
if exhausted_until is not None and now < exhausted_until:
Expand Down Expand Up @@ -1394,6 +1539,37 @@ def _is_suppressed(_p, _s): # type: ignore[misc]
},
)

elif provider == "xai-oauth":
# When the user logs in via ``hermes model`` -> xAI Grok OAuth,
# tokens are written to the auth.json singleton
# (``providers["xai-oauth"]``). Surface them in the pool too so
# ``hermes auth list`` reflects the logged-in state and so the pool
# is the single source of truth for refresh during runtime resolution.
if _is_suppressed(provider, "loopback_pkce"):
return changed, active_sources

state = _load_provider_state(auth_store, "xai-oauth")
tokens = state.get("tokens") if isinstance(state, dict) else None
if isinstance(tokens, dict) and tokens.get("access_token"):
active_sources.add("loopback_pkce")
from hermes_cli.auth import DEFAULT_XAI_OAUTH_BASE_URL

base_url = DEFAULT_XAI_OAUTH_BASE_URL
changed |= _upsert_entry(
entries,
provider,
"loopback_pkce",
{
"source": "loopback_pkce",
"auth_type": AUTH_TYPE_OAUTH,
"access_token": tokens.get("access_token", ""),
"refresh_token": tokens.get("refresh_token"),
"base_url": base_url,
"last_refresh": state.get("last_refresh"),
"label": label_from_token(tokens.get("access_token", ""), "loopback_pkce"),
},
)

return changed, active_sources


Expand Down
Loading
Loading