refactor(model): one persist writer for /model across CLI, gateway, TUI, dashboard; ACP + dashboard validate through switch_model

One `/model --global` produced four config.yaml shapes. CLI wrote
default/provider/base_url/api_mode and cleared the context pin on a route
change; the gateway rewrote the whole `model:` block (whole-file save_config)
and only set api_mode for `custom`; the TUI wrote three keys and never
touched api_mode, so a switch off an Anthropic-wire endpoint left a stale
`api_mode: anthropic_messages` in config; the dashboard main slot had its own
switched-provider logic, wrote `base_url: ""` and always dropped
context_length. ACP `session/set_model` and `POST /api/model/set` accepted
any model string (parse_model_input + detect_provider_for_model) so a model
no catalog knows, or a provider with no credentials, was handed to the
session / persisted and only failed at inference time.

Canonical: `hermes_cli.model_switch.model_selection_config_updates` (the
shape) + `persist_model_selection(result, config_path=None)` (targeted
per-key `atomic_roundtrip_yaml_update` writes, so sibling
`model_slots`/`model_fallback` keys survive; explicit path for the
multiplexed gateway's profile config) + `apply_model_selection` (same shape
applied to an in-memory `model:` dict for callers that save a whole
document). `atomic_roundtrip_yaml_update(value=None)` now REMOVES the key
instead of writing `key: null`, so per-key and whole-document writers land
the same file. Shape = CLI/gateway semantics: default, provider, base_url
(cleared when the target has none), api_mode (cleared when unresolved),
context_length cleared only when `should_clear_context_pin` says the route
identity changed, inline api_key/api cleared for non-custom targets.

Sites -> canonical:
  hermes_cli/cli_model_switch_mixin.py::_persist_global_switch          -> deleted; _commit_model_switch calls persist_model_selection
  hermes_cli/cli_model_switch_mixin.py::_clear_persisted_context_for_model_switch -> deleted (folded into the shape)
  gateway/slash_commands_model.py::_persist_model_switch_to_config       -> to_thread forwarder: persist_model_selection(result, ctx.config_path)
  tui_gateway/model_switch.py::_persist_model_switch                     -> deleted; _apply_model_switch calls persist_model_selection
  hermes_cli/web_server_config.py::_apply_main_model_assignment          -> apply_model_selection(result) (+ explicit custom api_key)
  hermes_cli/web_server_config.py::_validated_main_model_selection       -> NEW: switch_model(--provider) gate; rejection -> HTTP 400
  hermes_cli/web_routers/{models,profiles,config_env}.py main-slot paths -> through _validated_main_model_selection
  acp_adapter/server.py::_resolve_model_selection                        -> deleted; _switch_model calls switch_model (provider:model -> --provider), rejection -> ValueError

Behavior changes: TUI --global now writes/clears model.api_mode and clears a
route-changed context pin; gateway --global no longer rewrites the whole
model block (sibling keys survive) and clears api_mode for every target;
dashboard main slot / profile-create model / custom-endpoint activate now
reject unknown/uncredentialed/unlisted models (HTTP 400) and persist the
resolved base_url/api_mode instead of `base_url: ""`; ACP rejects the same
(ValueError surfaced by the command/protocol handler). Gateway persist runs
on a worker thread against the routed profile's config_path (multiplex-safe).
Cleared keys are removed from config.yaml rather than left as `null`. ACP
still never persists.

Kept `_normalize_main_model_assignment`: switch_model rejects a vendor name
posing as a provider (`moonshotai` -> "Unknown provider"), so the
vendor->aggregator repair is not a duplicate; E2E verified both branches.
No config migration: readers already coalesce `base_url: ""` to absent
(`_config_base_url_for_provider`) and gate api_mode on provider match
(`_provider_supports_explicit_api_mode`), so no stale-shape reader bug.

Tests: tests/hermes_cli/test_model_persist_one_shape.py (four surfaces land
one block; same-route re-pick keeps the pin), tests/acp_adapter/
test_acp_dashboard_model_switch_validation.py (rejection + explicit
provider prefix). Replaces test_acp_set_model_explicit_provider.py and the
two TUI-only persist tests; tests that intercepted the old per-surface seams
(`cli.save_config_value`, `load_config_readonly`, `tui_gateway.server.
_persist_model_switch`) now intercept the canonical seam. Each fix
sabotage-verified red.
This commit is contained in:
teknium1
2026-09-12 21:50:37 -07:00
committed by Teknium
parent 3e066dfedd
commit 11576390fe
21 changed files with 394 additions and 277 deletions
+22 -21
View File
@@ -305,32 +305,33 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
choice = encode_model_choice(provider, model)
return SessionModelState(available_models=[ModelInfo(model_id=choice, name=model)], current_model_id=choice)
@staticmethod
def _resolve_model_selection(raw_model: str, current_provider: str) -> tuple[str, str]:
"""Resolve ``provider:model`` input into the provider and normalized model id."""
target_provider, new_model = current_provider, raw_model.strip()
try:
from hermes_cli.models import detect_provider_for_model, parse_model_input
raw = new_model
target_provider, new_model = parse_model_input(new_model, current_provider)
# An explicit ``provider:model`` prefix is a selection; detection is a fallback for bare
# names only and must not second-guess it (#59089).
if target_provider == current_provider and new_model == raw:
detected = detect_provider_for_model(new_model, current_provider)
if detected:
target_provider, new_model = detected
except Exception:
logger.debug("Provider detection failed, using model as-is", exc_info=True)
return target_provider, new_model
def _switch_model(
self, state: SessionState, raw_model: str, *, keep_endpoint: bool = False
) -> tuple[str | None, str, str]:
"""Rebuild the session agent on a new model -> (old provider, new provider, model).
``keep_endpoint`` carries base_url/api_mode over when the provider is unchanged."""
Resolution goes through ``hermes_cli.model_switch.switch_model`` seeded with the live
agent route — the same catalog/alias/credential validation as CLI/gateway/TUI ``/model``
— so ACP never hands the session a model no provider can serve. ``provider:model`` picker
ids become ``--provider``. ACP never persists. ``keep_endpoint`` carries base_url/api_mode
over when the provider is unchanged."""
from hermes_cli.config import get_compatible_custom_providers, load_config
from hermes_cli.model_switch import switch_model
from hermes_cli.models import parse_model_input
current_provider = getattr(state.agent, "provider", None)
target_provider, new_model = self._resolve_model_selection(raw_model, current_provider or "openrouter")
explicit_provider, model_input = parse_model_input(raw_model, "")
cfg = load_config()
result = switch_model(
raw_input=model_input, explicit_provider=explicit_provider,
current_provider=current_provider or "openrouter", current_model=str(state.model or ""),
current_base_url=str(getattr(state.agent, "base_url", "") or ""),
current_api_key=str(getattr(state.agent, "api_key", "") or ""),
user_providers=cfg.get("providers") if isinstance(cfg.get("providers"), dict) else {},
custom_providers=get_compatible_custom_providers(cfg))
if not result.success:
raise ValueError(result.error_message or f"Cannot switch to {raw_model}")
target_provider, new_model = result.target_provider, result.new_model
state.model = new_model
endpoint: dict[str, Any] = {}
if keep_endpoint and not (current_provider and target_provider != current_provider):
+5 -41
View File
@@ -16,7 +16,7 @@ from typing import Any, Optional
from agent.i18n import t
from gateway.platforms.event import MessageEvent
from hermes_cli.config import atomic_config_write, clear_model_endpoint_credentials
from hermes_cli.config import atomic_config_write
from utils import base_url_host_matches
logger = logging.getLogger("gateway.run") # log-record parity with gateway/run.py
@@ -55,46 +55,10 @@ def _model_switch_skew_guard() -> Optional[str]:
async def _persist_model_switch_to_config(result, config_path) -> None:
"""Write-through a resolved /model switch to ``config_path`` (model.default/provider/base_url).
Raw read: merged defaults must not be persisted back. A scalar/None ``model:`` is coerced to a
dict first. Named providers re-resolve base_url/api_mode, so leftovers are cleared; custom
providers have no registry entry to re-derive from and need an explicit set-or-clear.
"""
from hermes_cli.config import read_user_config_raw, save_config
cfg = read_user_config_raw(config_path)
raw_model = cfg.get("model")
if isinstance(raw_model, dict):
model_cfg = raw_model
elif isinstance(raw_model, str) and raw_model.strip():
model_cfg = cfg["model"] = {"default": raw_model.strip()}
else:
model_cfg = cfg["model"] = {}
try:
from hermes_cli.route_identity import should_clear_context_pin_async
clear_pin = await should_clear_context_pin_async(
model_cfg.get("default") or model_cfg.get("model"), result.new_model,
model_cfg.get("base_url"), result.base_url, model_cfg.get("provider"), result.target_provider,
)
except Exception:
clear_pin = True
if clear_pin:
model_cfg.pop("context_length", None)
model_cfg["default"] = result.new_model
model_cfg["provider"] = result.target_provider
is_custom_target = str(result.target_provider or "").strip().lower() == "custom"
if result.base_url:
model_cfg["base_url"] = result.base_url
elif is_custom_target:
model_cfg.pop("base_url", None)
if not is_custom_target:
clear_model_endpoint_credentials(model_cfg, clear_base_url=True)
elif result.api_mode:
model_cfg["api_mode"] = result.api_mode
else:
model_cfg.pop("api_mode", None)
save_config(cfg)
"""Write-through a resolved /model switch to the profile config at ``config_path``, off the
event loop (the route comparison can do cold-start disk I/O)."""
from hermes_cli.model_switch import persist_model_selection
await asyncio.to_thread(persist_model_selection, result, config_path)
@dataclasses.dataclass
+2 -37
View File
@@ -170,7 +170,8 @@ def _commit_model_switch(
cli._pending_one_turn_model_restore = snapshot
_print_switch_summary(cli, result, old_model, one_turn=one_turn, strict_context=not picker)
if persist_global:
_persist_global_switch(cli, result)
from hermes_cli.model_switch import persist_model_selection
persist_model_selection(result)
_cprint(" Saved to config.yaml (--global)" if picker else " Saved to config.yaml")
elif one_turn:
_cprint(" (next turn only — restores after one response)")
@@ -182,24 +183,6 @@ def _commit_model_switch(
HermesCLI._persist_model_switch_to_session(cli, result)
def _persist_global_switch(cli, result) -> None:
"""Write the switched route to config.yaml (--global). base_url/api_mode are freshly resolved
for the target provider, so sync them every time (None clears a value the new provider doesn't
need) — otherwise the OLD provider's endpoint/wire-protocol lingers in config.yaml."""
from cli import HermesCLI, save_config_value
HermesCLI._clear_persisted_context_for_model_switch(cli, result)
save_config_value("model.default", result.new_model)
save_config_value("model.provider", result.target_provider)
# base_url/api_mode were previously never persisted here, so a global switch left the OLD provider's
# endpoint/wire-protocol in config.yaml. result.base_url/api_mode are always freshly resolved for the
# target provider (see model_switch.py), so sync them every time; None clears a value the new provider
# doesn't need (#25106).
# See _apply_model_switch_result above for why base_url/api_mode must be synced on every global switch
# (#25106).
save_config_value("model.base_url", result.base_url or None)
save_config_value("model.api_mode", result.api_mode or None)
def _show_model_picker(cli, ctx, force_refresh: bool) -> None:
"""``/model`` with no args: open the picker, or print usage when nothing is authed."""
from cli import _cprint
@@ -545,24 +528,6 @@ class CLIModelSwitchMixin:
scroll_offset = selected - visible + 1
return max(0, min(scroll_offset, n - visible)), visible
def _clear_persisted_context_for_model_switch(self, result) -> None:
"""Drop a global context pin when its configured owner changes."""
from cli import save_config_value
try:
from hermes_cli.config import load_config_readonly
from hermes_cli.route_identity import should_clear_context_pin
config = load_config_readonly()
model_cfg = config.get("model", {}) if isinstance(config, dict) else {}
if not isinstance(model_cfg, dict) or "context_length" not in model_cfg:
return
if should_clear_context_pin(
model_cfg.get("default") or model_cfg.get("model"), result.new_model,
model_cfg.get("base_url"), result.base_url,
model_cfg.get("provider"), result.target_provider):
save_config_value("model.context_length", None)
except Exception:
save_config_value("model.context_length", None)
def _stage_and_swap_model(self, result, old_model) -> bool:
"""Stage ``result`` onto the CLI fields, then swap the live agent in place.
+60
View File
@@ -1519,6 +1519,66 @@ def switch_model(
return _build_switch_result(st)
def model_selection_config_updates(result: ModelSwitchResult, current_model_cfg: Any) -> dict[str, Any]:
"""The ONE config.yaml shape a persisted model selection produces, as ``model.<key>`` -> value
(``None`` = clear). ``current_model_cfg`` is the on-disk ``model:`` block (raw).
base_url/api_mode are freshly resolved for the target route, so they are always synced —
``None`` when the target has none — otherwise the OLD provider's endpoint/wire-protocol lingers
(#25106). A context pin is dropped only when its route identity changed (fail-closed).
Non-custom targets resolve credentials from env/auth.json/the pool, so an inline
``model.api_key`` is a leftover that would contaminate later custom resolution."""
model_cfg = current_model_cfg if isinstance(current_model_cfg, dict) else {}
updates: dict[str, Any] = {
"default": result.new_model, "provider": result.target_provider,
"base_url": result.base_url or None, "api_mode": result.api_mode or None,
}
if "context_length" in model_cfg:
from hermes_cli.route_identity import should_clear_context_pin
if should_clear_context_pin(
model_cfg.get("default") or model_cfg.get("model"), result.new_model,
model_cfg.get("base_url"), result.base_url, model_cfg.get("provider"), result.target_provider):
updates["context_length"] = None
if not str(result.target_provider or "").strip().lower().startswith("custom"):
for key in ("api_key", "api"):
if key in model_cfg:
updates[key] = None
return updates
def apply_model_selection(model_cfg: Any, result: ModelSwitchResult) -> dict:
"""Apply the canonical shape to an in-memory ``model:`` dict (``None`` = key removed) for
callers that save a whole config document they are already mutating."""
model_cfg = dict(model_cfg) if isinstance(model_cfg, dict) else {}
for key, value in model_selection_config_updates(result, model_cfg).items():
if value is None:
model_cfg.pop(key, None)
else:
model_cfg[key] = value
return model_cfg
def persist_model_selection(result: ModelSwitchResult, config_path: Any = None) -> None:
"""Write a successful :func:`switch_model` result to ``config_path`` (default:
``HERMES_HOME/config.yaml`` — the context override or ``HERMES_HOME`` at call time).
Targeted key writes, not a whole-``model:`` rewrite: a block rewrite destroys sibling keys the
user set there (``model_slots``, ``model_fallback``, ...). ``should_clear_context_pin`` can do
cold-start disk I/O — async callers run this on a worker thread."""
from pathlib import Path
from hermes_cli.config import get_config_path, read_user_config_raw, warn_unpinned_cron_jobs_after_model_config_change
from utils import atomic_roundtrip_yaml_update
path = Path(config_path) if config_path else get_config_path()
for key, value in model_selection_config_updates(result, read_user_config_raw(path).get("model")).items():
atomic_roundtrip_yaml_update(path, f"model.{key}", value)
# Same unpinned-cron notice as `hermes config set` for every model switch.
warn_unpinned_cron_jobs_after_model_config_change(f"model.{key}", value)
try: # owner-only: config files contain API keys
os.chmod(path, 0o600)
except (OSError, NotImplementedError):
pass
def _extra_headers_from_config(entry: Any) -> dict[str, str]:
if not isinstance(entry, dict):
return {}
+5 -4
View File
@@ -15,6 +15,7 @@ from hermes_cli.web_routers._common import http_failure, scoped_to_thread
from hermes_cli.web_deps import LateState, late
from hermes_cli.web_server_config import (
_apply_main_model_assignment, _denormalize_config_from_web, _normalize_config_for_web, _schema_with_dynamic_provider_options,
_validated_main_model_selection,
)
from hermes_cli.web_server_profiles import (
_approval_mode_of, _broadcast_gateway_session_info, _is_other_profile, _parse_model_ids,
@@ -516,9 +517,8 @@ def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> T
cfg["providers"] = providers
if body.make_default:
cfg["model"] = _apply_main_model_assignment(
cfg.get("model", {}), endpoint_id, model, base_url
)
result = _validated_main_model_selection(cfg, endpoint_id, model, base_url)
cfg["model"] = _apply_main_model_assignment(cfg.get("model", {}), result)
if entry.get("key_env") and isinstance(cfg["model"], dict):
cfg["model"]["key_env"] = entry["key_env"]
cfg["model"].pop("api_key", None)
@@ -573,7 +573,8 @@ def activate_custom_endpoint(endpoint_id: str, profile: Optional[str] = None):
if not model or not base_url:
raise HTTPException(status_code=400, detail="custom endpoint is incomplete")
model_cfg = _apply_main_model_assignment(cfg.get("model", {}), provider_key, model, base_url)
model_cfg = _apply_main_model_assignment(
cfg.get("model", {}), _validated_main_model_selection(cfg, provider_key, model, base_url))
if entry.get("key_env"):
model_cfg["key_env"] = entry["key_env"]
model_cfg.pop("api_key", None)
+7 -4
View File
@@ -28,7 +28,9 @@ from typing import Any, Callable, Dict, List, Optional, Tuple
from fastapi import APIRouter, HTTPException, Query
from hermes_cli.web_deps import late
from hermes_cli.web_server_config import _apply_main_model_assignment, _normalize_main_model_assignment
from hermes_cli.web_server_config import (
_apply_main_model_assignment, _normalize_main_model_assignment, _validated_main_model_selection,
)
from hermes_cli.web_server_gateway import _strip_session_list_rows
from hermes_cli.web_server_profiles import (
_fallback_profile_dicts, _hub_action_name, _write_profile_mcp_servers,
@@ -96,13 +98,14 @@ def _profile_setup_command(name: str) -> str:
def _write_profile_model(profile_dir: Path, provider: str, model: str) -> None:
"""Write the main model assignment into ``profile_dir``'s config.yaml (HERMES_HOME-scoped);
clears stale ``base_url`` / ``context_length`` like ``POST /api/model/set`` does."""
"""Write the main model assignment into ``profile_dir``'s config.yaml (HERMES_HOME-scoped)
through the same validated /model shape as ``POST /api/model/set``."""
from hermes_cli.config import load_config, save_config
with _hermes_home_scope(profile_dir):
provider, model = _normalize_main_model_assignment(provider, model)
cfg = load_config()
cfg["model"] = _apply_main_model_assignment(cfg.get("model", {}), provider, model)
result = _validated_main_model_selection(cfg, provider, model)
cfg["model"] = _apply_main_model_assignment(cfg.get("model", {}), result)
save_config(cfg)
+42 -36
View File
@@ -4,7 +4,7 @@
import logging
import os
from fastapi import HTTPException
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, List, Optional, Tuple, TYPE_CHECKING
from agent.model_metadata import is_local_endpoint
from hermes_cli.config import (
DEFAULT_CONFIG,
@@ -17,6 +17,9 @@ from hermes_cli.config import (
)
from hermes_cli.web_server_memory import _normalize_memory_provider_name
if TYPE_CHECKING:
from hermes_cli.model_switch import ModelSwitchResult
# Same logger the code used before extraction (record parity).
_log = logging.getLogger("hermes_cli.web_server")
@@ -442,43 +445,43 @@ def _normalize_main_model_assignment(provider: str, model: str) -> tuple[str, st
return prov_in, model_in
def _apply_main_model_assignment(
model_cfg: "Any", provider: str, model: str, base_url: str = "", api_key: str = ""
) -> dict:
"""Apply a main-slot model assignment to a ``model`` config dict in place.
def _validated_main_model_selection(
cfg: dict, provider: str, model: str, base_url: str = "", api_key: str = ""
) -> "ModelSwitchResult":
"""Route a dashboard main-slot pick through ``switch_model`` (catalog/alias/credential
validation) seeded with the configured route, exactly like a ``/model <model> --provider
<provider> --global``. A bare ``custom`` target carries the submitted endpoint as the current
one, which is how ``switch_model`` binds a custom base_url/key. Rejections become 400s."""
from hermes_cli.config import get_compatible_custom_providers
from hermes_cli.model_switch import switch_model
Sets ``provider``/``default``, then reconciles endpoint fields. ``base_url`` and the
endpoint key share one lifecycle: an explicit value is always persisted; an existing
value is cleared ONLY when switching to a *different* provider (it belonged to the old
endpoint); a same-provider re-pick preserves it — re-picking a model used to wipe a
user's custom host (e.g. a Xiaomi MiMo Token Plan URL) and break their keys. The
runtime resolver reads ``model.base_url`` from config and only honors it when the
configured provider matches, so preserving it here is what lets the override route.
A stale secret may live under the legacy ``api`` alias with no ``api_key``, so the
switch-clears-the-key path triggers on either field. ``context_length`` is always
dropped (the new model may have a different window).
model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {}
is_bare_custom = provider.strip().lower() in {"custom", "local"}
result = switch_model(
raw_input=model, explicit_provider=provider, is_global=True,
current_provider=str(model_cfg.get("provider") or ""), current_model=str(model_cfg.get("default") or ""),
current_base_url=base_url if is_bare_custom else str(model_cfg.get("base_url") or ""),
current_api_key=api_key if is_bare_custom else "",
user_providers=cfg.get("providers") if isinstance(cfg.get("providers"), dict) else {},
custom_providers=get_compatible_custom_providers(cfg))
if not result.success:
raise HTTPException(status_code=400, detail=result.error_message or "model switch rejected")
return result
Returns the same dict (a fresh dict if the input wasn't one).
"""
if not isinstance(model_cfg, dict):
model_cfg = {}
prev_provider = str(model_cfg.get("provider") or "").strip().lower()
new_provider = provider.strip().lower()
switched = new_provider != prev_provider
model_cfg["provider"] = provider
model_cfg["default"] = model
if base_url.strip():
model_cfg["base_url"] = base_url.strip()
elif model_cfg.get("base_url") and switched:
model_cfg["base_url"] = ""
def _apply_main_model_assignment(model_cfg: "Any", result: "ModelSwitchResult", api_key: str = "") -> dict:
"""Apply a main-slot selection to a ``model`` config dict via the canonical /model shape
(``hermes_cli.model_switch.apply_model_selection``). An explicit key for a custom endpoint is
the one inline credential the runtime reads (``model.api_key``); the legacy ``api`` alias is
dropped so a stale secret cannot shadow it.
Returns a new dict."""
from hermes_cli.model_switch import apply_model_selection
model_cfg = apply_model_selection(model_cfg, result)
if api_key.strip():
model_cfg["api_key"] = api_key.strip()
model_cfg.pop("api", None)
elif (model_cfg.get("api_key") or model_cfg.get("api")) and switched:
clear_model_endpoint_credentials(model_cfg, clear_api_mode=False)
if switched:
clear_model_endpoint_credentials(model_cfg, clear_api_key=False)
model_cfg.pop("context_length", None)
return model_cfg
@@ -656,7 +659,9 @@ def _apply_main_assignment_sync(cfg: dict, provider: str, model: str, base_url:
provider_entry = providers_cfg.get(provider) if isinstance(providers_cfg, dict) else None
if not base_url and isinstance(provider_entry, dict) and provider_entry.get("base_url"):
base_url = str(provider_entry.get("base_url") or "").strip()
model_cfg = _apply_main_model_assignment(cfg.get("model", {}), provider, model, base_url, api_key)
result = _validated_main_model_selection(cfg, provider, model, base_url, api_key)
provider, model = result.target_provider, result.new_model
model_cfg = _apply_main_model_assignment(cfg.get("model", {}), result, api_key)
_resolve_assignment_credentials(model_cfg, provider, provider_entry)
cfg["model"] = model_cfg
@@ -826,8 +831,9 @@ def _denormalize_config_from_web(config: Dict[str, Any]) -> Dict[str, Any]:
new_provider, resolved_model = _infer_provider_on_model_change(model_val, prev_provider)
if new_provider and new_provider.strip().lower() != prev_provider.lower():
norm_provider, norm_model = _normalize_main_model_assignment(new_provider, resolved_model)
disk_model = _apply_main_model_assignment(disk_model, norm_provider, norm_model)
model_val = norm_model
result = _validated_main_model_selection(load_config(), norm_provider, norm_model)
disk_model = _apply_main_model_assignment(disk_model, result)
model_val = result.new_model
disk_model["default"] = model_val
if ctx_sent:
if ctx_override > 0:
@@ -0,0 +1,71 @@
"""ACP ``session/set_model`` and the dashboard main slot validate through ``switch_model``.
Both surfaces used to accept any string (``parse_model_input`` + ``detect_provider_for_model``
for ACP; bare provider/model normalization for ``POST /api/model/set``), so a model no catalog
knew — or a provider with no credentials — was handed to the session / written to config.yaml
and only failed at inference time. They now share the CLI/gateway/TUI ``/model`` pipeline: a
rejection from ``switch_model`` is a rejection on these surfaces too, and an acceptance carries
the resolved (provider, model) — an explicit ``provider:model`` prefix is honoured as
``--provider`` (#59089), never re-detected.
"""
from __future__ import annotations
import types
import pytest
from hermes_cli.model_switch import ModelSwitchResult
def _acp_agent():
from acp_adapter.server import HermesACPAgent
made: dict = {}
class _SM:
def _make_agent(self, **kw):
made.update(kw)
return types.SimpleNamespace(provider=kw.get("requested_provider"), model=kw.get("model"))
def save_session(self, sid):
pass
return HermesACPAgent(session_manager=_SM()), made
def _state():
return types.SimpleNamespace(
session_id="s1", cwd=".", model="claude-sonnet-5",
agent=types.SimpleNamespace(provider="anthropic", base_url="https://api.anthropic.com", api_key="k"))
def test_acp_and_dashboard_reject_what_switch_model_rejects(monkeypatch):
rejected = ModelSwitchResult(success=False, error_message="Unknown provider 'notaprovider'.")
monkeypatch.setattr("hermes_cli.model_switch.switch_model", lambda **_kw: rejected)
agent, made = _acp_agent()
state = _state()
with pytest.raises(ValueError, match="Unknown provider"):
agent._switch_model(state, "notaprovider:whatever")
assert made == {} and state.model == "claude-sonnet-5" # session untouched
from fastapi import HTTPException
from hermes_cli.web_server_config import _apply_model_assignment_sync
with pytest.raises(HTTPException) as exc:
_apply_model_assignment_sync("main", "notaprovider", "whatever", "", "")
assert exc.value.status_code == 400 and "Unknown provider" in exc.value.detail
def test_acp_explicit_provider_prefix_becomes_explicit_provider(monkeypatch):
seen: dict = {}
def _switch(**kw):
seen.update(kw)
return ModelSwitchResult(success=True, new_model=kw["raw_input"], target_provider=kw["explicit_provider"])
monkeypatch.setattr("hermes_cli.model_switch.switch_model", _switch)
agent, made = _acp_agent()
old, new_provider, model = agent._switch_model(_state(), "anthropic:claude-sonnet-5", keep_endpoint=True)
assert (seen["explicit_provider"], seen["raw_input"]) == ("anthropic", "claude-sonnet-5")
assert (old, new_provider, model) == ("anthropic", "anthropic", "claude-sonnet-5")
assert made["requested_provider"] == "anthropic" and made["base_url"] == "https://api.anthropic.com"
@@ -1,28 +0,0 @@
"""ACP ``session/set_model``: an explicit ``provider:model`` prefix is a selection, never re-detected.
Regression for #59089: ``anthropic:claude-sonnet-5`` while already on anthropic resolved to the same
provider, so the bare-name fallback ``detect_provider_for_model`` ran and could hand the session to
OpenRouter because the bare name appears in its catalog.
"""
from __future__ import annotations
from acp_adapter.server import HermesACPAgent
def test_explicit_provider_prefix_skips_detection(monkeypatch):
calls: list[tuple[str, str]] = []
def hijack(model, current):
calls.append((model, current))
return ("openrouter", f"anthropic/{model}")
monkeypatch.setattr("hermes_cli.models.detect_provider_for_model", hijack)
assert HermesACPAgent._resolve_model_selection("anthropic:claude-sonnet-5", "anthropic") == (
"anthropic", "claude-sonnet-5")
assert calls == []
def test_bare_name_still_uses_detection(monkeypatch):
monkeypatch.setattr("hermes_cli.models.detect_provider_for_model", lambda m, c: ("deepseek", m))
assert HermesACPAgent._resolve_model_selection("deepseek-flash", "anthropic") == ("deepseek", "deepseek-flash")
@@ -102,7 +102,8 @@ async def test_model_global_persists_when_config_has_flat_string_model(tmp_path,
)
assert written["model"]["default"] == "gpt-5.5"
assert written["model"]["provider"] == "openrouter"
assert "base_url" not in written["model"]
# The resolved aggregator endpoint is persisted (same shape as CLI/TUI --global, #25106).
assert written["model"]["base_url"] == "https://openrouter.ai/api/v1"
@pytest.mark.asyncio
+3 -2
View File
@@ -195,9 +195,10 @@ async def test_picker_tap_global_flag_persists(tmp_path, monkeypatch, seed_model
)
assert written["model"]["default"] == "gpt-5.5"
assert written["model"]["provider"] == "openrouter"
assert "base_url" not in written["model"]
# The resolved aggregator endpoint is persisted (same shape as CLI/TUI --global, #25106).
assert written["model"]["base_url"] == "https://openrouter.ai/api/v1"
assert "api_key" not in written["model"]
assert "api_mode" not in written["model"]
assert written["model"]["api_mode"] == "chat_completions"
assert "context_length" not in written["model"]
@@ -8,9 +8,9 @@ Both ``_handle_model_switch`` (typed ``/model <name>``) and
``_apply_model_switch_result`` (interactive picker) shared the same gap: the
persistence block wrote ``model.default``/``model.provider`` but never
touched ``base_url``/``api_mode`` at all. Fix: sync both on every global
switch, clearing to ``None`` when the resolved result doesn't need them —
mirroring the already-correct ``tui_gateway/server.py:_persist_model_switch``
pattern (fixed for #48305).
switch, clearing to ``None`` when the resolved result doesn't need them — now the
canonical ``hermes_cli.model_switch.persist_model_selection`` shape shared by every
surface.
"""
from unittest.mock import MagicMock, patch
@@ -80,10 +80,10 @@ def _run_switch(monkeypatch, result, cmd="/model MiniMax-M3 --global"):
monkeypatch.setattr(cli_mod, "_cprint", lambda *a, **k: None)
saved: dict[str, object] = {}
def _fake_save(key, value):
def _fake_save(path, key, value):
saved[key] = value
monkeypatch.setattr(cli_mod, "save_config_value", _fake_save)
monkeypatch.setattr("utils.atomic_roundtrip_yaml_update", _fake_save)
monkeypatch.setattr("hermes_cli.model_switch.switch_model", lambda **kw: result)
monkeypatch.setattr(
"hermes_cli.inventory.load_picker_context",
@@ -105,13 +105,13 @@ def test_global_switch_persists_base_url_and_api_mode(monkeypatch):
def test_session_only_switch_does_not_touch_config(monkeypatch):
"""--session must not call save_config_value at all — persistence stays
"""--session must not write config.yaml at all — persistence stays
entirely in-memory."""
import cli as cli_mod
monkeypatch.setattr(cli_mod, "_cprint", lambda *a, **k: None)
save_calls = []
monkeypatch.setattr(cli_mod, "save_config_value", lambda *a, **k: save_calls.append(a))
monkeypatch.setattr("utils.atomic_roundtrip_yaml_update", lambda *a, **k: save_calls.append(a))
monkeypatch.setattr("hermes_cli.model_switch.switch_model", lambda **kw: _make_result())
monkeypatch.setattr(
"hermes_cli.inventory.load_picker_context",
@@ -134,10 +134,10 @@ def _run_apply(monkeypatch, result, persist_global=True):
monkeypatch.setattr(cli_mod, "_cprint", lambda *a, **k: None)
saved: dict[str, object] = {}
def _fake_save(key, value):
def _fake_save(path, key, value):
saved[key] = value
monkeypatch.setattr(cli_mod, "save_config_value", _fake_save)
monkeypatch.setattr("utils.atomic_roundtrip_yaml_update", _fake_save)
cli_mod.HermesCLI._apply_model_switch_result(_StubCLI(), result, persist_global)
return saved
@@ -131,9 +131,8 @@ def test_global_switch_clears_context_pin_owned_by_previous_route(monkeypatch):
writes = []
monkeypatch.setattr(cli_mod, "_cprint", lambda *_a, **_k: None)
monkeypatch.setattr(
cli_mod,
"save_config_value",
lambda key, value: writes.append((key, value)),
"utils.atomic_roundtrip_yaml_update",
lambda path, key, value: writes.append((key, value)),
)
cli = _StubCLI()
cli.model = "shared-model"
@@ -169,7 +168,7 @@ def test_global_switch_clears_context_pin_owned_by_previous_route(monkeypatch):
"agent.model_metadata.get_model_context_length",
return_value=256_000,
),
patch("hermes_cli.config.load_config_readonly", return_value=configured),
patch("hermes_cli.config.read_user_config_raw", return_value=configured),
):
cli_mod.HermesCLI._apply_model_switch_result(cli, result, True)
@@ -0,0 +1,127 @@
"""One config.yaml shape for a persisted model selection, whichever surface wrote it.
``/model --global`` reaches config.yaml from four places (CLI mixin, gateway slash command, TUI
gateway, dashboard main slot). Each used to hand-roll its own write; the TUI never touched
``api_mode`` (stale wire protocol after a switch) and the dashboard wrote ``base_url: ""``. All
four now go through ``hermes_cli.model_switch.persist_model_selection`` /
``apply_model_selection``, so the same ``ModelSwitchResult`` must land as the same ``model.*``
keys on disk — including the api_mode clear and the route-changed context_length clear.
"""
from __future__ import annotations
import asyncio
import pytest
import yaml
from hermes_cli.model_switch import ModelSwitchResult
_SEED = (
"model:\n"
" default: local-model\n"
" provider: custom\n"
" base_url: http://localhost:1234/v1\n"
" api_mode: anthropic_messages\n"
" api_key: sk-stale\n"
" context_length: 32000\n"
" model_slots:\n"
" fast: gpt-5-mini\n"
"agent:\n"
" system_prompt: keepme\n"
)
_RESULT = ModelSwitchResult(
success=True, new_model="claude-haiku", target_provider="anthropic", provider_changed=True,
api_key="sk-new", base_url="", api_mode="", is_global=True,
)
@pytest.fixture
def seeded_home(tmp_path, monkeypatch):
import cli
(tmp_path / "config.yaml").write_text(_SEED, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setattr(cli, "_hermes_home", tmp_path)
return tmp_path
def _model_block(home) -> dict:
return yaml.safe_load((home / "config.yaml").read_text(encoding="utf-8"))["model"]
def _via_cli(home):
from hermes_cli import cli_model_switch_mixin as mixin
stub = type("Stub", (), {
"agent": None, "model": "local-model", "_pending_one_turn_model_restore": None,
"_stage_and_swap_model": lambda self, r, o: True})()
mixin._commit_model_switch(stub, _RESULT, persist_global=True)
def _via_gateway(home):
from gateway.slash_commands_model import _persist_model_switch_to_config
asyncio.run(_persist_model_switch_to_config(_RESULT, home / "config.yaml"))
def _via_tui(home):
from tui_gateway import server
session = {"agent": None}
server._apply_model_switch(
"sid", session, "claude-haiku --provider anthropic --global", confirm_expensive_model=True)
def _via_dashboard(home):
from hermes_cli.web_server_config import _apply_model_assignment_sync
_apply_model_assignment_sync("main", "anthropic", "claude-haiku", "", "")
@pytest.mark.parametrize("surface", [_via_cli, _via_gateway, _via_tui, _via_dashboard],
ids=["cli", "gateway", "tui", "dashboard"])
def test_every_persist_surface_writes_the_same_model_block(seeded_home, monkeypatch, surface):
monkeypatch.setattr("hermes_cli.model_switch.switch_model", lambda **_kw: _RESULT)
monkeypatch.setattr("cli.HermesCLI._persist_model_switch_to_session", lambda *a, **k: None)
monkeypatch.setattr("hermes_cli.cli_model_switch_mixin._print_switch_summary", lambda *a, **k: None)
monkeypatch.setattr("hermes_cli.model_selection_guards.combined_selection_warning",
lambda *a, **k: None, raising=False)
monkeypatch.setattr("cli._cprint", lambda *a, **k: None, raising=False)
surface(seeded_home)
block = _model_block(seeded_home)
assert (block["default"], block["provider"]) == ("claude-haiku", "anthropic")
# The target route has no endpoint/wire override: the OLD custom ones must not linger.
assert not block.get("base_url") and not block.get("api_mode")
# Route identity changed → the pinned window belongs to the old route. Cleared keys are
# REMOVED, not left as ``key: null`` litter.
assert "context_length" not in block and "base_url" not in block and "api_mode" not in block
# Non-custom providers never read an inline key; the stale secret is gone.
assert "api_key" not in block
# Sibling keys under ``model:`` survive (targeted writes, no block rewrite).
assert block["model_slots"] == {"fast": "gpt-5-mini"}
assert yaml.safe_load((seeded_home / "config.yaml").read_text())["agent"]["system_prompt"] == "keepme"
def test_same_route_repick_keeps_the_context_pin(seeded_home):
"""A model re-pick on the SAME route keeps ``context_length`` (only the owner change drops it)."""
from hermes_cli.model_switch import persist_model_selection
same_route = ModelSwitchResult(
success=True, new_model="local-model", target_provider="custom",
base_url="http://localhost:1234/v1", api_mode="anthropic_messages", is_global=True)
persist_model_selection(same_route)
block = _model_block(seeded_home)
assert block["context_length"] == 32000
assert block["api_key"] == "sk-stale" # custom targets keep their inline key
def test_gateway_persists_to_the_profile_config_it_was_given(tmp_path, monkeypatch):
"""Multiplexed gateway: the write lands in the routed profile's config.yaml, never the
process-level HERMES_HOME."""
from gateway.slash_commands_model import _persist_model_switch_to_config
process_home, profile_home = tmp_path / "default", tmp_path / "profiles" / "named"
for home in (process_home, profile_home):
home.mkdir(parents=True)
(home / "config.yaml").write_text("model:\n default: old\n provider: openai-codex\n")
monkeypatch.setenv("HERMES_HOME", str(process_home))
asyncio.run(_persist_model_switch_to_config(_RESULT, profile_home / "config.yaml"))
assert _model_block(profile_home)["default"] == "claude-haiku"
assert _model_block(process_home)["default"] == "old"
+3 -2
View File
@@ -3285,8 +3285,9 @@ class TestDenormalizeProviderSwitch:
model = result["model"]
assert model["provider"] == "openrouter"
assert model["default"] == "google/gemini-2.5-flash"
# The old ollama-local endpoint must not carry over to openrouter.
assert not model.get("base_url")
# The old ollama-local endpoint must not carry over to openrouter (the switch resolves
# the aggregator's own endpoint instead of leaving the field blank or stale).
assert model.get("base_url") != "http://localhost:11434/v1"
def test_context_length_override_survives_provider_switch(self):
@@ -229,6 +229,18 @@ class TestProfileScopedMcp:
class TestProfileScopedModel:
@pytest.fixture(autouse=True)
def _accept_any_model(self, monkeypatch):
"""These tests pin WHICH profile the write lands in, not catalog validation: the main
slot now routes through ``switch_model`` (needs credentials + a listed model), so echo the
request back as an accepted route."""
from hermes_cli.model_switch import ModelSwitchResult
def _switch(*, raw_input, explicit_provider, **_kw):
return ModelSwitchResult(success=True, new_model=raw_input, target_provider=explicit_provider)
monkeypatch.setattr("hermes_cli.model_switch.switch_model", _switch)
def test_model_set_main_scoped(self, client, isolated_profiles):
resp = client.post(
"/api/model/set",
+3 -74
View File
@@ -4423,7 +4423,7 @@ def test_apply_model_switch_persist_override_false_never_persists(monkeypatch):
lambda *a: pytest.fail("persist_override must bypass resolve_persist_behavior"),
)
monkeypatch.setattr(
server, "_persist_model_switch",
"hermes_cli.model_switch.persist_model_selection",
lambda _r: pytest.fail("persist_override=False must not persist"),
)
monkeypatch.setattr(
@@ -9646,9 +9646,9 @@ def test_config_set_model_global_persists(monkeypatch):
monkeypatch.setattr("hermes_cli.model_switch.switch_model", _switch_model)
monkeypatch.setattr(server, "_restart_slash_worker", lambda sid, session: None)
monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
# _persist_model_switch uses targeted save_config_value writes (#48305) so it
# persist_model_selection uses targeted per-key writes (#48305) so it
# preserves sibling model.* keys instead of rewriting the whole block.
monkeypatch.setattr("cli.save_config_value", lambda key, value: saved_values.__setitem__(key, value) or True)
monkeypatch.setattr("utils.atomic_roundtrip_yaml_update", lambda path, key, value: saved_values.__setitem__(key, value))
resp = server.handle_request(
{
@@ -19852,77 +19852,6 @@ def test_get_usage_safe_when_active_count_raises(monkeypatch):
assert usage["model"] == "x"
def test_persist_model_switch_preserves_sibling_model_keys(tmp_path, monkeypatch):
"""#48305: switching models from the TUI must NOT destroy sibling keys under
`model:` (model_slots, model_fallback, etc.). _persist_model_switch now uses
targeted save_config_value writes instead of rewriting the whole block."""
import types
import yaml
import cli
cfg_path = tmp_path / "config.yaml"
cfg_path.write_text(
"model:\n"
" default: old-model\n"
" provider: openai\n"
" model_slots:\n"
" fast: gpt-5-mini\n"
" model_fallback:\n"
" - claude-haiku\n"
"agent:\n"
" system_prompt: keepme\n"
)
# save_config_value() resolves the config path from get_hermes_home() (live
# env var), always targeting HERMES_HOME/config.yaml — point it at tmp_path.
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setattr(cli, "_hermes_home", tmp_path)
result = types.SimpleNamespace(
new_model="new-model", target_provider="anthropic", base_url=None
)
server._persist_model_switch(result)
saved = yaml.safe_load(cfg_path.read_text())
# The switched fields updated...
assert saved["model"]["default"] == "new-model"
assert saved["model"]["provider"] == "anthropic"
# ...and the sibling keys SURVIVED (the bug was that they got wiped).
assert saved["model"]["model_slots"] == {"fast": "gpt-5-mini"}
assert saved["model"]["model_fallback"] == ["claude-haiku"]
assert saved["agent"]["system_prompt"] == "keepme"
def test_persist_model_switch_clears_stale_base_url(tmp_path, monkeypatch):
"""#48305: switching from a custom endpoint (which set model.base_url) to a
provider with no base_url must CLEAR the stale base_url, not leave it
pointing at the old host."""
import types
import yaml
import cli
cfg_path = tmp_path / "config.yaml"
cfg_path.write_text(
"model:\n"
" default: local-model\n"
" provider: custom:mylocal\n"
" base_url: http://localhost:1234/v1\n"
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setattr(cli, "_hermes_home", tmp_path)
# Switch to a native provider with no base_url.
result = types.SimpleNamespace(
new_model="claude-haiku", target_provider="anthropic", base_url=None
)
server._persist_model_switch(result)
saved = yaml.safe_load(cfg_path.read_text())
assert saved["model"]["default"] == "claude-haiku"
assert saved["model"]["provider"] == "anthropic"
# Stale custom base_url must be cleared (null coalesces to absent on read).
assert not saved["model"].get("base_url"), saved["model"].get("base_url")
# ---------------------------------------------------------------------------
# _resolve_runtime_with_fallback — init-time provider fallback
# ---------------------------------------------------------------------------
@@ -121,7 +121,7 @@ def test_apply_model_switch_does_not_leak_process_env():
patch("tui_gateway.server._emit"),
patch("tui_gateway.server._restart_slash_worker"),
patch("tui_gateway.server._session_info", return_value={}),
patch("tui_gateway.server._persist_model_switch") as mock_persist,
patch("hermes_cli.model_switch.persist_model_selection") as mock_persist,
):
before = {k: os.environ.get(k) for k in env_keys}
result = server._apply_model_switch("sidB", sess_b, "glm-5.1")
@@ -35,6 +35,13 @@ def home(tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
# The profile write now validates through ``switch_model`` (catalog + credentials); these
# tests pin the guard handshake, so echo the pick back as an accepted route.
from hermes_cli.model_switch import ModelSwitchResult
monkeypatch.setattr(
"hermes_cli.model_switch.switch_model",
lambda *, raw_input, explicit_provider, **_kw: ModelSwitchResult(
success=True, new_model=raw_input, target_provider=explicit_provider))
return hermes_home
+2 -12
View File
@@ -11,17 +11,6 @@ from .method_ctx import HandlerRegistry, bind_module
_registry = HandlerRegistry()
def _persist_model_switch(result) -> None:
# Targeted key writes: a full `model:` block rewrite via save_config() would destroy
# sibling keys the user set there (`model_slots`, `model_fallback`, ...).
from cli import save_config_value
save_config_value("model.default", result.new_model)
save_config_value("model.provider", result.target_provider)
# A provider without a base_url must clear the stale one (custom endpoint -> native)
# or the new model routes at the old host; reads coalesce null to absent.
save_config_value("model.base_url", result.base_url or None)
_RUNTIME_KEYS = ("model", "provider", "api_key", "base_url", "api_mode")
@@ -240,7 +229,8 @@ def _apply_model_switch(
"model": result.new_model, "provider": result.target_provider,
"base_url": result.base_url, "api_key": result.api_key, "api_mode": result.api_mode}
if persist_global:
_persist_model_switch(result)
from hermes_cli.model_switch import persist_model_selection
persist_model_selection(result)
return {
"value": result.new_model, "warning": result.warning_message or "",
"confirm_required": False,
+8 -1
View File
@@ -400,6 +400,8 @@ def atomic_roundtrip_yaml_update(path: Union[str, Path], key_path: str, value: A
Narrower than :func:`atomic_yaml_write` on purpose: for user-edited config files where a
single setting mutation must not disturb the rest. Still writes via temp file + atomic replace.
``value=None`` removes the key (a ``key: null`` leftover reads as absent everywhere but
litters the file and diverges from whole-document writers that drop the key).
"""
from ruamel.yaml.comments import CommentedMap
# Honor escaped dots and prefer existing literal dotted keys (model IDs like ``glm-5.3``) over
@@ -418,10 +420,15 @@ def atomic_roundtrip_yaml_update(path: Union[str, Path], key_path: str, value: A
remaining = keys[i:]
seg, consumed = _greedy_literal_match(dict(current), remaining) or (remaining[0], 1)
if i + consumed == len(keys):
current[seg] = value
if value is None:
current.pop(seg, None)
else:
current[seg] = value
break
next_value = current.get(seg)
if not isinstance(next_value, CommentedMap):
if value is None:
return # nothing to remove under a missing/scalar parent
next_value = CommentedMap()
current[seg] = next_value
current = next_value