Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
177 changes: 176 additions & 1 deletion tests/test_tui_gateway_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
from pathlib import Path
from unittest.mock import patch

import pytest

from hermes_constants import reset_hermes_home_override, set_hermes_home_override
from hermes_cli.active_sessions import active_session_registry_snapshot
from tui_gateway import server
Expand Down Expand Up @@ -1098,7 +1100,7 @@ def fake_agent(**kwargs):
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda requested=None, target_model=None: {
lambda **kwargs: {
"provider": "openai-codex",
"base_url": "https://chatgpt.com/backend-api/codex",
"api_key": "token",
Expand All @@ -1117,6 +1119,179 @@ def fake_agent(**kwargs):
assert captured["platform"] == "tui"


def test_resolve_runtime_with_auth_fallback_returns_fallback_entry_model(monkeypatch):
"""Helper returns (runtime_dict, config entry model str) — not the chain list."""
from hermes_cli.auth import AuthError

fallback_chain = [
{"provider": "openrouter", "model": "meta-llama/llama-4-maverick"},
]
fallback_runtime = {
"provider": "openrouter",
"base_url": "https://openrouter.ai/api/v1",
"api_key": "fallback-key",
"api_mode": "openai_chat",
}

def fake_resolve(**kwargs):
requested = kwargs.get("requested")
if requested in (None, "xai-oauth"):
raise AuthError("xAI OAuth state is missing access_token.")
return fallback_runtime

monkeypatch.setattr(server, "_load_fallback_model", lambda: fallback_chain)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
fake_resolve,
)

runtime, fb_model = server._resolve_runtime_with_auth_fallback()

assert fb_model == "meta-llama/llama-4-maverick"
assert runtime == fallback_runtime


def test_make_agent_falls_back_on_primary_auth_error(monkeypatch):
"""Desktop must not hard-error when primary OAuth is dead but fallback works (#43588)."""
from hermes_cli.auth import AuthError

captured = {}
fallback_chain = [
{"provider": "openrouter", "model": "meta-llama/llama-4-maverick"},
]
fallback_runtime = {
"provider": "openrouter",
"base_url": "https://openrouter.ai/api/v1",
"api_key": "fallback-key",
"api_mode": "openai_chat",
"credential_pool": None,
}
calls = []

def fake_resolve(**kwargs):
requested = kwargs.get("requested")
calls.append(requested)
if requested in (None, "xai-oauth"):
raise AuthError(
"xAI OAuth state is missing access_token. Re-authenticate with `hermes model`."
)
return dict(fallback_runtime)

def fake_agent(**kwargs):
captured.update(kwargs)
return types.SimpleNamespace(model=kwargs.get("model"))

monkeypatch.delenv("HERMES_MODEL", raising=False)
monkeypatch.delenv("HERMES_INFERENCE_MODEL", raising=False)
monkeypatch.delenv("HERMES_TUI_PROVIDER", raising=False)
monkeypatch.setattr(
server,
"_load_cfg",
lambda: {
"model": {"default": "grok-3", "provider": "xai-oauth"},
"fallback_providers": fallback_chain,
},
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
fake_resolve,
)
monkeypatch.setattr("run_agent.AIAgent", fake_agent)
monkeypatch.setattr(server, "_load_enabled_toolsets", lambda: ["file"])
monkeypatch.setattr(server, "_get_db", lambda: None)

agent = server._make_agent("sid", "session-key")

assert calls[:2] == [None, "openrouter"]
# fb_model string becomes the agent's active model.
assert agent.model == "meta-llama/llama-4-maverick"
assert captured["model"] == "meta-llama/llama-4-maverick"
assert captured["provider"] == fallback_runtime["provider"]
assert captured["api_key"] == fallback_runtime["api_key"]
# Mid-turn fallback chain is still passed separately via _load_fallback_model().
assert captured["fallback_model"] == fallback_chain


def test_make_agent_auth_fallback_skips_stale_session_overrides(monkeypatch):
"""Resumed session overrides from a dead primary must not poison fallback creds."""
from hermes_cli.auth import AuthError

captured = {}
fallback_runtime = {
"provider": "openrouter",
"base_url": "https://openrouter.ai/api/v1",
"api_key": "fallback-key",
"api_mode": "openai_chat",
}

def fake_resolve(**kwargs):
requested = kwargs.get("requested")
if requested in (None, "xai-oauth"):
raise AuthError("xAI OAuth state is missing access_token.")
return dict(fallback_runtime)

def fake_agent(**kwargs):
captured.update(kwargs)
return types.SimpleNamespace(model=kwargs.get("model"))

monkeypatch.setattr(
server,
"_load_cfg",
lambda: {
"fallback_providers": [
{"provider": "openrouter", "model": "meta-llama/llama-4-maverick"},
],
},
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
fake_resolve,
)
monkeypatch.setattr("run_agent.AIAgent", fake_agent)
monkeypatch.setattr(server, "_load_enabled_toolsets", lambda: [])
monkeypatch.setattr(server, "_get_db", lambda: None)

server._make_agent(
"sid",
"session-key",
model_override={
"model": "grok-3",
"provider": "xai-oauth",
"base_url": "https://api.x.ai/v1",
"api_key": "stale-oauth-token",
},
)

assert captured["provider"] == fallback_runtime["provider"]
assert captured["api_key"] == fallback_runtime["api_key"]
assert captured["base_url"] == fallback_runtime["base_url"]


def test_make_agent_auth_error_without_fallback_raises(monkeypatch):
from hermes_cli.auth import AuthError

monkeypatch.delenv("HERMES_MODEL", raising=False)
monkeypatch.delenv("HERMES_INFERENCE_MODEL", raising=False)
monkeypatch.delenv("HERMES_TUI_PROVIDER", raising=False)
monkeypatch.setattr(
server,
"_load_cfg",
lambda: {"model": {"default": "grok-3", "provider": "xai-oauth"}},
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda **kwargs: (_ for _ in ()).throw(
AuthError("xAI OAuth state is missing access_token.")
),
)
monkeypatch.setattr("run_agent.AIAgent", lambda **kwargs: types.SimpleNamespace())
monkeypatch.setattr(server, "_load_enabled_toolsets", lambda: [])
monkeypatch.setattr(server, "_get_db", lambda: None)

with pytest.raises(RuntimeError, match="access_token"):
server._make_agent("sid", "session-key")


def test_background_agent_kwargs_preserves_full_fallback_chain(monkeypatch):
chain = [
{"provider": "openrouter", "model": "openai/gpt-5.5"},
Expand Down
110 changes: 98 additions & 12 deletions tui_gateway/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -2807,6 +2807,86 @@ def _load_fallback_model():
return get_fallback_chain(_load_cfg())


def _resolve_runtime_with_auth_fallback(
*,
requested: str | None = None,
target_model: str | None = None,
explicit_base_url: str | None = None,
explicit_api_key: str | None = None,
) -> tuple[dict, str | None]:
"""Resolve provider credentials; try ``fallback_providers`` on ``AuthError``.

Mirrors gateway ``_resolve_runtime_agent_kwargs`` / CLI
``_ensure_runtime_credentials`` so Desktop does not hard-error on a dead
primary OAuth provider when a configured fallback chain is healthy.

Returns ``(runtime_dict, fallback_model_or_none)``. When the second value
is set, callers should use it as the agent model.
"""
from hermes_cli.auth import AuthError, is_rate_limited_auth_error
from hermes_cli.runtime_provider import (
format_runtime_provider_error,
resolve_runtime_provider,
)

try:
runtime = resolve_runtime_provider(
requested=requested,
target_model=target_model,
explicit_base_url=explicit_base_url,
explicit_api_key=explicit_api_key,
)
return runtime, None
except AuthError as auth_exc:
if is_rate_limited_auth_error(auth_exc):
logger.warning(
"Primary provider rate-limited (429): %s — trying fallback",
auth_exc,
)
else:
logger.warning(
"Primary provider auth failed: %s — trying fallback",
auth_exc,
)

fb_list = _load_fallback_model()
if not fb_list:
raise RuntimeError(format_runtime_provider_error(auth_exc)) from auth_exc

for entry in fb_list:
try:
fb_provider = entry.get("provider")
fb_model = entry.get("model")
fb_api_key = entry.get("api_key")
if not fb_api_key:
key_env = str(
entry.get("key_env") or entry.get("api_key_env") or ""
).strip()
if key_env:
fb_api_key = os.getenv(key_env, "").strip() or None
runtime = resolve_runtime_provider(
requested=fb_provider,
target_model=fb_model,
explicit_base_url=entry.get("base_url"),
explicit_api_key=fb_api_key,
)
logger.info(
"Fallback provider resolved: %s model=%s",
fb_provider or runtime.get("provider"),
fb_model,
)
return runtime, fb_model
except Exception as fb_exc:
logger.debug(
"Fallback entry %s failed: %s",
entry.get("provider"),
fb_exc,
)
continue

raise RuntimeError(format_runtime_provider_error(auth_exc)) from auth_exc


def _agent_fallback_model(agent):
"""Return an agent's fallback chain without rehydrating deliberately empty chains."""
if hasattr(agent, "_fallback_chain"):
Expand Down Expand Up @@ -3024,7 +3104,6 @@ def _make_agent(
service_tier_override: str | None = None,
):
from run_agent import AIAgent
from hermes_cli.runtime_provider import resolve_runtime_provider

# MCP tool discovery runs in a background daemon thread at startup so a
# dead server can't freeze the shell (see tui_gateway/entry.py). The agent
Expand Down Expand Up @@ -3065,29 +3144,36 @@ def _make_agent(
override_base_url = model_override.get("base_url")
override_api_key = model_override.get("api_key")
override_api_mode = model_override.get("api_mode")
runtime = resolve_runtime_provider(
runtime, fallback_model = _resolve_runtime_with_auth_fallback(
requested=requested_provider,
target_model=model or None,
)
# The switch already resolved concrete credentials/endpoint; honor them
# so a custom/named endpoint survives the rebuild even if global
# resolution would pick a different one.
if override_base_url:
runtime["base_url"] = override_base_url
if override_api_key:
runtime["api_key"] = override_api_key
if override_api_mode:
runtime["api_mode"] = override_api_mode
if fallback_model:
model = fallback_model
elif override_base_url or override_api_key or override_api_mode:
# The switch already resolved concrete credentials/endpoint; honor them
# so a custom/named endpoint survives the rebuild even if global
# resolution would pick a different one. Skip when we fell back —
# stale persisted overrides from a dead primary must not clobber
# healthy fallback credentials.
if override_base_url:
runtime["base_url"] = override_base_url
if override_api_key:
runtime["api_key"] = override_api_key
if override_api_mode:
runtime["api_mode"] = override_api_mode
else:
model, requested_provider = _resolve_startup_runtime()
if isinstance(model_override, str) and model_override:
model = model_override
if provider_override:
requested_provider = provider_override
runtime = resolve_runtime_provider(
runtime, fallback_model = _resolve_runtime_with_auth_fallback(
requested=requested_provider,
target_model=model or None,
)
if fallback_model:
model = fallback_model
return AIAgent(
model=model,
max_iterations=_cfg_max_turns(cfg, 90),
Expand Down