fix(auxiliary): resolve xai oauth compression from pool

This commit is contained in:
helix4u 2026-05-15 18:43:39 -06:00 committed by Teknium
parent c7db6a5800
commit 97a32afdc4
3 changed files with 119 additions and 10 deletions

View file

@ -1272,12 +1272,40 @@ def _resolve_nous_runtime_api(*, force_refresh: bool = False) -> Optional[tuple[
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).
Prefer the credential pool, matching the main runtime/provider status
path. Some xAI OAuth logins live only as pool entries; falling straight
to the singleton auth-store resolver would make auxiliary tasks such as
compression report "no provider configured" even though ``hermes auth
status`` shows xAI OAuth as logged in.
Falls back to ``hermes_cli.auth``'s singleton runtime resolver for older
auth-store-only logins. Returns ``None`` if the user is not authenticated
with xAI Grok OAuth.
"""
try:
from hermes_cli.auth import DEFAULT_XAI_OAUTH_BASE_URL
pool = load_pool("xai-oauth")
if pool and pool.has_credentials():
entry = pool.select()
if entry is not None:
api_key = str(
getattr(entry, "runtime_api_key", None)
or getattr(entry, "access_token", "")
or ""
).strip()
base_url = str(
os.getenv("HERMES_XAI_BASE_URL", "").strip().rstrip("/")
or os.getenv("XAI_BASE_URL", "").strip().rstrip("/")
or getattr(entry, "runtime_base_url", None)
or getattr(entry, "base_url", None)
or DEFAULT_XAI_OAUTH_BASE_URL
).strip().rstrip("/")
if api_key and base_url:
return api_key, base_url
except Exception as exc:
logger.debug("Auxiliary xAI OAuth pool credential resolution failed: %s", exc)
try:
from hermes_cli.auth import resolve_xai_oauth_runtime_credentials

View file

@ -3237,11 +3237,20 @@ class AIAgent:
except Exception:
_aux_cfg_provider = ""
if client is None or not aux_model:
msg = (
"⚠ No auxiliary LLM provider configured — context "
"compression will drop middle turns without a summary. "
"Run `hermes setup` or set OPENROUTER_API_KEY."
)
if _aux_cfg_provider and _aux_cfg_provider != "auto":
msg = (
"⚠ Configured auxiliary compression provider "
f"'{_aux_cfg_provider}' is unavailable — context "
"compression will drop middle turns without a summary. "
"Check auxiliary.compression in config.yaml and "
"reauthenticate that provider."
)
else:
msg = (
"⚠ No auxiliary LLM provider configured — context "
"compression will drop middle turns without a summary. "
"Run `hermes setup` or set OPENROUTER_API_KEY."
)
self._compression_warning = msg
self._emit_status(msg)
logger.warning(

View file

@ -26,6 +26,7 @@ from agent.auxiliary_client import (
_normalize_aux_provider,
_try_payment_fallback,
_resolve_auto,
_resolve_xai_oauth_for_aux,
_CodexCompletionsAdapter,
)
@ -221,6 +222,77 @@ class TestReadCodexAccessToken:
assert result == "plain-token-no-jwt"
class TestResolveXaiOAuthForAux:
def test_uses_pool_backed_credentials_without_singleton(self, tmp_path, monkeypatch):
"""Auxiliary xAI OAuth must see pool-only credentials.
``hermes auth status`` already reports these as logged in; compression
should not fall through to "no auxiliary provider configured" just
because the singleton auth-store entry is absent.
"""
from agent.credential_pool import AUTH_TYPE_OAUTH, PooledCredential, load_pool
from hermes_cli.auth import DEFAULT_XAI_OAUTH_BASE_URL
hermes_home = tmp_path / "hermes"
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "auth.json").write_text(json.dumps({
"version": 1,
"providers": {},
}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.delenv("HERMES_XAI_BASE_URL", raising=False)
monkeypatch.delenv("XAI_BASE_URL", raising=False)
pool = load_pool("xai-oauth")
pool.add_entry(PooledCredential(
provider="xai-oauth",
id="xai123",
label="pool-only",
auth_type=AUTH_TYPE_OAUTH,
priority=0,
source="manual:xai_pkce",
access_token="pool-access-token",
refresh_token="pool-refresh-token",
base_url=DEFAULT_XAI_OAUTH_BASE_URL,
))
assert _resolve_xai_oauth_for_aux() == (
"pool-access-token",
DEFAULT_XAI_OAUTH_BASE_URL,
)
def test_pool_backed_credentials_honor_base_url_env_override(self, tmp_path, monkeypatch):
from agent.credential_pool import AUTH_TYPE_OAUTH, PooledCredential, load_pool
from hermes_cli.auth import DEFAULT_XAI_OAUTH_BASE_URL
hermes_home = tmp_path / "hermes"
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "auth.json").write_text(json.dumps({
"version": 1,
"providers": {},
}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("HERMES_XAI_BASE_URL", "https://example.x.ai/v1/")
pool = load_pool("xai-oauth")
pool.add_entry(PooledCredential(
provider="xai-oauth",
id="xai456",
label="pool-only",
auth_type=AUTH_TYPE_OAUTH,
priority=0,
source="manual:xai_pkce",
access_token="pool-access-token",
refresh_token="pool-refresh-token",
base_url=DEFAULT_XAI_OAUTH_BASE_URL,
))
assert _resolve_xai_oauth_for_aux() == (
"pool-access-token",
"https://example.x.ai/v1",
)
class TestAnthropicOAuthFlag:
"""Test that OAuth tokens get is_oauth=True in auxiliary Anthropic client."""