mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-21 16:18:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

fix(auth): preserve fallback routes and OAuth state

Browse source 
Switch provider and model together after setup-time auth failure. Serialize global auth-store merges under target-specific locks and preserve auth-to-shared lock ordering for profile OAuth refreshes.
This commit is contained in:
Dan Schnurbusch 2026-07-15 18:28:25 -05:00 • committed by Teknium
parent 261b0f8240
commit f68fd80f41
8 changed files with 388 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

14
agent/credential_pool.py
View file

@ -543,14 +543,12 @@ def _write_through_provider_state_to_global_root(
except Exception:
return
try:
if global_path.exists():
global_store = _load_auth_store(global_path)
else:
global_store = {}
if not isinstance(global_store, dict):
return
_store_provider_state(global_store, provider_id, dict(state), set_active=False)
auth_mod._save_auth_store(global_store, global_path)
auth_mod._persist_provider_state_to_store(
provider_id,
state,
global_path,
set_active=False,
)
except Exception as exc: # pragma: no cover - best effort
logger.debug(
"%s pool refresh: write-through to global root failed: %s",

55
cron/scheduler.py
View file

@ -2988,13 +2988,9 @@ def run_job(
except Exception:
pass
# Reasoning config from config.yaml (per-model override > global) —
# resolved through the shared chokepoint against the job's effective
# model (per-job override > HERMES_MODEL env > config.yaml default).
# Reasoning config is resolved after provider authentication so an auth
# fallback can first replace the primary model with its configured model.
from hermes_constants import resolve_reasoning_config
reasoning_config = resolve_reasoning_config(
_cfg if isinstance(_cfg, dict) else {}, str(model)
)
# Prefill messages from env or config.yaml. The top-level
# prefill_messages_file key is canonical; agent.prefill_messages_file is
@ -3040,6 +3036,8 @@ def run_job(
# off-host call is ever made with a stored key.
_guard_job_credential_exfil(job)
primary_model_for_drift = model
primary_provider_for_drift = (job.get("provider") or "").strip().lower() or None
try:
# Do not inject HERMES_INFERENCE_PROVIDER here. resolve_runtime_provider()
# already prefers persisted config over stale shell/env overrides when
@ -3048,33 +3046,64 @@ def run_job(
# example DeepSeek) for cron jobs that do not pin provider/model.
runtime_kwargs = {
"requested": job.get("provider"),
"target_model": model or None,
}
if job.get("base_url"):
runtime_kwargs["explicit_base_url"] = job.get("base_url")
runtime = resolve_runtime_provider(**runtime_kwargs)
primary_provider_for_drift = (
str(runtime.get("provider") or "").strip().lower()
or primary_provider_for_drift
)
except AuthError as auth_exc:
# Primary provider auth failed — try fallback chain before giving up.
# Primary provider auth failed — try each configured provider/model
# pair atomically. Keeping the primary model while changing only the
# provider can silently route a paid GPT model through OpenRouter.
primary_provider_for_drift = (
str(getattr(auth_exc, "provider", "") or "").strip().lower()
or primary_provider_for_drift
)
logger.warning("Job '%s': primary auth failed (%s), trying fallback", job_id, auth_exc)
fb_list = get_fallback_chain(_cfg)
runtime = None
for entry in fb_list:
if not isinstance(entry, dict):
continue
fb_provider = str(entry.get("provider") or "").strip()
if not fb_provider:
continue
fb_model = str(entry.get("model") or "").strip() or None
try:
fb_kwargs = {"requested": entry.get("provider")}
fb_kwargs = {
"requested": fb_provider,
"target_model": fb_model,
}
if entry.get("base_url"):
fb_kwargs["explicit_base_url"] = entry["base_url"]
if entry.get("api_key"):
fb_kwargs["explicit_api_key"] = entry["api_key"]
runtime = resolve_runtime_provider(**fb_kwargs)
logger.info("Job '%s': fallback resolved to %s", job_id, runtime.get("provider"))
if fb_model:
model = fb_model
logger.info(
"Job '%s': fallback resolved to %s%s",
job_id,
runtime.get("provider"),
f" model {fb_model}" if fb_model else "",
)
break
except Exception as fb_exc:
logger.debug("Job '%s': fallback %s failed: %s", job_id, entry.get("provider"), fb_exc)
logger.debug("Job '%s': fallback %s failed: %s", job_id, fb_provider, fb_exc)
if runtime is None:
raise RuntimeError(format_runtime_provider_error(auth_exc)) from auth_exc
except Exception as exc:
message = format_runtime_provider_error(exc)
raise RuntimeError(message) from exc
reasoning_config = resolve_reasoning_config(
_cfg if isinstance(_cfg, dict) else {}, str(model)
)
# Provider/model-drift fail-closed guard (#44585).
#
# An UNPINNED job (no explicit job["provider"]/["model"]) follows the
@ -3097,14 +3126,16 @@ def run_job(
_drift: list[str] = []
_provider_snapshot = (job.get("provider_snapshot") or "").strip().lower()
if _provider_snapshot and not (job.get("provider") or "").strip():
_current_provider = str(runtime.get("provider") or "").strip().lower()
_current_provider = str(
primary_provider_for_drift or runtime.get("provider") or ""
).strip().lower()
if _current_provider and _current_provider != _provider_snapshot:
_drift.append(
f"provider '{_provider_snapshot}' -> '{_current_provider}'"
)
_model_snapshot = (job.get("model_snapshot") or "").strip().lower()
if _model_snapshot and not (job.get("model") or "").strip():
_current_model = str(model or "").strip().lower()
_current_model = str(primary_model_for_drift or "").strip().lower()
if _current_model and _current_model != _model_snapshot:
_drift.append(
f"model '{_model_snapshot}' -> '{_current_model}'"

129
hermes_cli/auth.py
View file

@ -985,6 +985,28 @@ def _auth_lock_path() -> Path:
_auth_lock_holder = threading.local()
_auth_target_lock_holders: Dict[str, threading.local] = {}
_auth_target_lock_holders_guard = threading.Lock()
def _same_path(left: Path, right: Path) -> bool:
try:
return left.resolve(strict=False) == right.resolve(strict=False)
except Exception:
return left == right
def _auth_lock_holder_for(target_path: Path) -> threading.local:
"""Return a reentrancy tracker scoped to one auth-store lock path."""
active_path = _auth_file_path()
if _same_path(target_path, active_path):
return _auth_lock_holder
try:
key = str(target_path.resolve(strict=False))
except Exception:
key = str(target_path)
with _auth_target_lock_holders_guard:
return _auth_target_lock_holders.setdefault(key, threading.local())
@contextmanager
@ -1060,8 +1082,16 @@ def _file_lock(
@contextmanager
def _auth_store_lock(timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS):
"""Cross-process advisory lock for auth.json reads+writes. Reentrant.
def _auth_store_lock(
timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS,
*,
target_path: Optional[Path] = None,
):
"""Cross-process advisory lock for one auth.json read/write transaction.
``target_path`` is required for profile-to-global write-throughs. A profile
lock does not protect the distinct global auth store; each path therefore
uses its own reentrancy tracker and kernel lock.
Lock ordering invariant: when this lock is held together with
``_nous_shared_store_lock``, acquire ``_auth_store_lock`` FIRST
@ -1069,9 +1099,11 @@ def _auth_store_lock(timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS):
refresh paths follow this order; violating it risks deadlock
against a concurrent import on the shared store.
"""
auth_path = target_path if target_path is not None else _auth_file_path()
lock_path = auth_path.with_suffix(".lock") if target_path is not None else _auth_lock_path()
with _file_lock(
_auth_lock_path(),
_auth_lock_holder,
lock_path,
_auth_lock_holder_for(auth_path),
timeout_seconds,
"Timed out waiting for auth store lock",
):
@ -1204,6 +1236,37 @@ def _load_provider_state_with_source(
return None, None
@contextmanager
def _provider_state_transaction(provider_id: str):
"""Lock the active auth store and any global fallback source in order.
Profile-backed refresh paths must take the global auth-store lock before
any provider-specific shared-store lock. Re-reading the source after the
target lock is acquired prevents both stale refreshes and whole-file lost
updates without inverting the documented auth -> shared lock order.
"""
with _auth_store_lock():
auth_store = _load_auth_store()
state, source_path = _load_provider_state_with_source(
auth_store,
provider_id,
)
active_path = _auth_file_path()
if source_path is None or _same_path(source_path, active_path):
yield auth_store, state, source_path
return
with _auth_store_lock(target_path=source_path):
source_store = _load_auth_store(source_path)
source_providers = source_store.get("providers")
source_state = None
if isinstance(source_providers, dict):
raw_state = source_providers.get(provider_id)
if isinstance(raw_state, dict):
source_state = dict(raw_state)
yield auth_store, source_state, source_path
def _load_provider_state(auth_store: Dict[str, Any], provider_id: str) -> Optional[Dict[str, Any]]:
"""Return a provider's persisted state.
@ -1247,9 +1310,12 @@ def _save_provider_state_to_source(
_save_auth_store(auth_store)
return
source_store = _load_auth_store(source_path)
_save_provider_state(source_store, provider_id, state)
_save_auth_store(source_store, target_path=source_path)
_persist_provider_state_to_store(
provider_id,
state,
source_path,
set_active=True,
)
def _store_provider_state(
@ -1268,6 +1334,25 @@ def _store_provider_state(
auth_store["active_provider"] = provider_id
def _persist_provider_state_to_store(
provider_id: str,
state: Dict[str, Any],
target_path: Path,
*,
set_active: bool = False,
) -> Path:
"""Merge one provider into a specific auth store under that store's lock."""
with _auth_store_lock(target_path=target_path):
auth_store = _load_auth_store(target_path)
_store_provider_state(
auth_store,
provider_id,
dict(state),
set_active=set_active,
)
return _save_auth_store(auth_store, target_path=target_path)
def mark_provider_active_if_unset(provider_id: str) -> None:
"""Set ``active_provider`` to *provider_id* only when none is set yet.
@ -4009,14 +4094,12 @@ def _write_through_xai_oauth_to_global_root(state: Dict[str, Any]) -> None:
except Exception:
return
try:
if global_path.exists():
global_store = _load_auth_store(global_path)
else:
global_store = {}
if not isinstance(global_store, dict):
return
_store_provider_state(global_store, "xai-oauth", dict(state), set_active=False)
_save_auth_store(global_store, global_path)
_persist_provider_state_to_store(
"xai-oauth",
state,
global_path,
set_active=False,
)
except Exception as exc: # pragma: no cover - best effort
logger.debug("xAI OAuth: write-through to global root failed: %s", exc)
@ -5240,9 +5323,11 @@ def resolve_nous_access_token(
refresh_skew_seconds: int = ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
) -> str:
"""Resolve a refresh-aware Nous Portal access token for managed tool gateways."""
with _auth_store_lock():
auth_store = _load_auth_store()
state, state_source_path = _load_provider_state_with_source(auth_store, "nous")
with _provider_state_transaction("nous") as (
auth_store,
state,
state_source_path,
):
if not state:
raise AuthError(
@ -5571,9 +5656,11 @@ def resolve_nous_runtime_credentials(
"""
sequence_id = uuid.uuid4().hex[:12]
with _auth_store_lock():
auth_store = _load_auth_store()
state, state_source_path = _load_provider_state_with_source(auth_store, "nous")
with _provider_state_transaction("nous") as (
auth_store,
state,
state_source_path,
):
if not state:
raise AuthError("Hermes is not logged into Nous Portal.",

82
tests/agent/test_credential_pool_oauth_writethrough.py
View file

@ -17,6 +17,7 @@ mocking the save boundary, so they exercise the actual atomic write path.
"""
import json
import threading
import pytest
@ -190,6 +191,87 @@ def test_write_through_helper_is_noop_in_classic_mode(monkeypatch, tmp_path):
)
def test_global_write_through_preserves_concurrent_root_update(
profile_and_root, monkeypatch
):
"""A stale profile write-through must not erase a concurrent root login."""
_profile_path, root_path = profile_and_root
_write_store(
root_path,
{
"version": 1,
"providers": {
"xai-oauth": {
"tokens": {"access_token": "old-xai", "refresh_token": "old-r"}
}
},
"credential_pool": {
"anthropic": [{"id": "anthropic-existing"}],
"openrouter": [{"id": "openrouter-existing"}],
},
},
)
helper_loaded = threading.Event()
allow_helper_save = threading.Event()
writer_started = threading.Event()
writer_done = threading.Event()
real_auth_load = A._load_auth_store
def paused_helper_load(path=None):
store = real_auth_load(path)
if threading.current_thread().name == "profile-write-through":
helper_loaded.set()
assert allow_helper_save.wait(timeout=5)
return store
monkeypatch.setattr(A, "_load_auth_store", paused_helper_load)
def profile_write_through():
CP._write_through_provider_state_to_global_root(
"xai-oauth",
{"tokens": {"access_token": "new-xai", "refresh_token": "new-r"}},
)
def concurrent_codex_login():
writer_started.set()
with A._auth_store_lock(target_path=root_path):
store = A._load_auth_store(root_path)
A._store_provider_state(
store,
"openai-codex",
{"tokens": {"access_token": "codex-a", "refresh_token": "codex-r"}},
set_active=False,
)
pool = store.setdefault("credential_pool", {})
pool["openai-codex"] = [{"id": "codex-login"}]
A._save_auth_store(store, target_path=root_path)
writer_done.set()
helper = threading.Thread(target=profile_write_through, name="profile-write-through")
helper.start()
assert helper_loaded.wait(timeout=5)
writer = threading.Thread(target=concurrent_codex_login, name="concurrent-login")
writer.start()
assert writer_started.wait(timeout=5)
# Before the fix the writer completes while the stale helper is paused.
# After the fix it blocks on the root lock until the helper saves and exits.
writer_done.wait(timeout=0.2)
allow_helper_save.set()
helper.join(timeout=5)
writer.join(timeout=5)
assert not helper.is_alive()
assert not writer.is_alive()
root = _read_store(root_path)
assert root["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "new-r"
assert root["providers"]["openai-codex"]["tokens"]["refresh_token"] == "codex-r"
assert root["credential_pool"]["openai-codex"] == [{"id": "codex-login"}]
assert root["credential_pool"]["anthropic"] == [{"id": "anthropic-existing"}]
assert root["credential_pool"]["openrouter"] == [{"id": "openrouter-existing"}]
def test_codex_pool_refresh_holds_auth_store_lock_across_post(monkeypatch, tmp_path):
"""The Codex OAuth pool refresh must POST under the cross-process auth lock.

51
tests/cron/test_scheduler.py
View file

@ -2002,6 +2002,57 @@ class TestRunJobConfigEnvVarExpansion:
"config.yaml ${VAR} in fallback_providers was not expanded."
)
def test_auth_fallback_switches_provider_and_model_together(self, tmp_path):
"""Codex auth failure must produce OpenRouter+GLM, never OpenRouter+GPT."""
from hermes_cli.auth import AuthError
(tmp_path / "config.yaml").write_text(
"model:\n"
" default: gpt-5.6-sol\n"
" provider: openai-codex\n"
"fallback_providers:\n"
" - provider: openrouter\n"
" model: z-ai/glm-5.2\n",
encoding="utf-8",
)
job = {
"id": "auth-fallback",
"name": "auth fallback",
"prompt": "hi",
"provider_snapshot": "openai-codex",
"model_snapshot": "gpt-5.6-sol",
}
fake_db = MagicMock()
def resolve_runtime(**kwargs):
if kwargs.get("requested") in (None, "openai-codex"):
raise AuthError(
"No Codex credentials stored", provider="openai-codex"
)
assert kwargs["requested"] == "openrouter"
assert kwargs["target_model"] == "z-ai/glm-5.2"
return {**self._RUNTIME, "provider": "openrouter"}
with patch("cron.scheduler._hermes_home", tmp_path), \
patch("cron.scheduler._resolve_origin", return_value=None), \
patch("hermes_cli.env_loader.load_hermes_dotenv"), \
patch("hermes_cli.env_loader.reset_secret_source_cache"), \
patch("hermes_state.SessionDB", return_value=fake_db), \
patch("hermes_cli.runtime_provider.resolve_runtime_provider",
side_effect=resolve_runtime), \
patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), \
patch("run_agent.AIAgent") as mock_agent_cls:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "ok"}
mock_agent_cls.return_value = mock_agent
success, _, _, error = run_job(job)
assert success is True
assert error is None
kwargs = mock_agent_cls.call_args.kwargs
assert kwargs["provider"] == "openrouter"
assert kwargs["model"] == "z-ai/glm-5.2"
def test_fallback_chain_merges_providers_and_legacy_model(self, tmp_path, monkeypatch):
"""Cron uses get_fallback_chain so legacy fallback_model is not dropped."""
(tmp_path / "config.yaml").write_text(

40
tests/hermes_cli/test_auth_profile_fallback.py
View file

@ -12,6 +12,7 @@ authenticated only at the global root.
from __future__ import annotations
import json
from contextlib import contextmanager
from pathlib import Path
import pytest
@ -450,3 +451,42 @@ def test_write_credential_pool_targets_profile_not_global(profile_env):
# Subsequent read returns profile (shadows global).
assert [e["id"] for e in read_credential_pool("openrouter")] == ["prof-new"]
def test_provider_state_transaction_locks_global_fallback_before_use(
profile_env,
monkeypatch,
):
"""Profile refreshes lock the root source before provider-specific locks."""
import hermes_cli.auth as auth
_write(
profile_env["global"] / "auth.json",
_make_auth_store(providers={"nous": {"access_token": "global-token"}}),
)
_write(profile_env["profile"] / "auth.json", _make_auth_store(providers={}))
entered = []
real_file_lock = auth._file_lock
@contextmanager
def recording_file_lock(lock_path, holder, timeout_seconds, timeout_message):
entered.append(lock_path)
with real_file_lock(
lock_path,
holder,
timeout_seconds,
timeout_message,
):
yield
monkeypatch.setattr(auth, "_file_lock", recording_file_lock)
with auth._provider_state_transaction("nous") as (_store, state, source):
assert state == {"access_token": "global-token"}
assert source == profile_env["global"] / "auth.json"
assert entered[:2] == [
profile_env["profile"] / "auth.lock",
profile_env["global"] / "auth.lock",
]

40
tests/test_tui_gateway_server.py
View file

@ -9528,8 +9528,11 @@ class TestResolveRuntimeWithFallback:
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda **kw: expected,
)
result = server._resolve_runtime_with_fallback({"requested": "openai"})
assert result == expected
runtime, fallback_model = server._resolve_runtime_with_fallback(
{"requested": "openai"}
)
assert runtime == expected
assert fallback_model is None
def test_auth_error_tries_fallback_chain(self, monkeypatch):
"""On AuthError from primary, walk fallback_providers chain."""
@ -9551,10 +9554,11 @@ class TestResolveRuntimeWithFallback:
"_load_fallback_model",
lambda: [{"provider": "deepseek", "model": "deepseek-v4-pro"}],
)
result = server._resolve_runtime_with_fallback(
runtime, fallback_model = server._resolve_runtime_with_fallback(
{"requested": "openai-codex"},
)
assert result == fallback_runtime
assert runtime == fallback_runtime
assert fallback_model == "deepseek-v4-pro"
def test_auth_error_all_fallbacks_fail_raises(self, monkeypatch):
"""When all fallbacks also fail, re-raise the original AuthError."""
@ -9602,10 +9606,11 @@ class TestResolveRuntimeWithFallback:
{"provider": "anthropic", "model": "claude-sonnet-4-6"},
],
)
result = server._resolve_runtime_with_fallback(
runtime, fallback_model = server._resolve_runtime_with_fallback(
{"requested": "openai-codex"},
)
assert result == fallback_runtime
assert runtime == fallback_runtime
assert fallback_model == "claude-sonnet-4-6"
def test_make_agent_uses_fallback_on_auth_error(self, monkeypatch):
"""Integration: _make_agent falls back to configured fallback
@ -9615,10 +9620,14 @@ class TestResolveRuntimeWithFallback:
from hermes_cli.auth import AuthError
captured = {}
fallback_runtime = {"provider": "deepseek", "api_key": "fb-tok"}
fallback_runtime = {
"provider": "deepseek",
"api_key": "fb-tok",
"base_url": "https://fallback.invalid/v1",
}
def fake_resolve(**kwargs):
if kwargs.get("requested") == "openai-codex":
if kwargs.get("requested") in (None, "openai-codex"):
raise AuthError("No Codex credentials stored")
return fallback_runtime
@ -9647,10 +9656,21 @@ class TestResolveRuntimeWithFallback:
monkeypatch.setattr(server, "_load_enabled_toolsets", lambda: ["file"])
monkeypatch.setattr(server, "_get_db", lambda: None)
agent = server._make_agent("sid", "session-key")
agent = server._make_agent(
"sid",
"session-key",
model_override={
"model": "gpt-5.5",
"provider": "openai-codex",
"base_url": "https://chatgpt.com/backend-api/codex",
"api_key": "stale-codex-token",
},
)
assert agent.model == "gpt-5.5"
assert agent.model == "deepseek-v4-pro"
assert captured["provider"] == "deepseek"
assert captured["base_url"] == "https://fallback.invalid/v1"
assert captured["api_key"] == "fb-tok"
def test_get_usage_does_not_substitute_cumulative_total_for_context_used():

48
tui_gateway/server.py
View file

@ -4473,20 +4473,19 @@ def _schedule_mcp_late_refresh(sid: str, agent) -> None:
def _resolve_runtime_with_fallback(
resolve_kwargs: dict | None = None,
) -> dict:
"""Resolve runtime provider with init-time fallback on auth failure.
) -> tuple[dict, str | None]:
"""Resolve a runtime and the fallback model selected after auth failure.
Mirrors the fallback pattern in ``cron/scheduler.py`` and
``hermes_cli/cli_agent_setup_mixin.py``: when the primary provider
raises ``AuthError``, walk the configured ``fallback_providers`` /
``fallback_model`` chain before giving up.
A fallback entry is one provider/model pair. Returning the model alongside
the runtime prevents callers from accidentally pairing the fallback
provider with the unavailable primary model.
"""
from hermes_cli.auth import AuthError
from hermes_cli.runtime_provider import resolve_runtime_provider
kwargs = resolve_kwargs or {}
try:
return resolve_runtime_provider(**kwargs)
return resolve_runtime_provider(**kwargs), None
except AuthError as primary_exc:
fb_chain = _load_fallback_model() or []
for entry in fb_chain:
@ -4495,8 +4494,11 @@ def _resolve_runtime_with_fallback(
fb_provider = (entry.get("provider") or "").strip()
if not fb_provider:
continue
fb_model = (entry.get("model") or "").strip() or None
try:
fb_kwargs: dict = {"requested": fb_provider}
if fb_model:
fb_kwargs["target_model"] = fb_model
if entry.get("base_url"):
fb_kwargs["explicit_base_url"] = entry["base_url"]
if entry.get("api_key"):
@ -4505,11 +4507,12 @@ def _resolve_runtime_with_fallback(
import logging
logging.getLogger(__name__).warning(
"Primary auth failed (%s), falling back to %s",
"Primary auth failed (%s), falling back to %s%s",
primary_exc,
fb_provider,
f" model {fb_model}" if fb_model else "",
)
return runtime
return runtime, fb_model
except Exception:
continue
raise
@ -4610,26 +4613,31 @@ def _make_agent(
resolve_kwargs["explicit_base_url"] = override_base_url
resolve_kwargs["requested"] = requested_provider
resolve_kwargs["target_model"] = model or None
runtime = _resolve_runtime_with_fallback(resolve_kwargs)
# 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
runtime, auth_fallback_model = _resolve_runtime_with_fallback(resolve_kwargs)
if auth_fallback_model:
model = auth_fallback_model
else:
# The switch already resolved concrete credentials/endpoint; honor
# persisted overrides only while using that original runtime. They
# must not leak into a different fallback provider/model pair.
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_with_fallback({
runtime, auth_fallback_model = _resolve_runtime_with_fallback({
"requested": requested_provider,
"target_model": model or None,
})
if auth_fallback_model:
model = auth_fallback_model
_pr = _load_provider_routing()
return AIAgent(
model=model,