mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Victor Kyriazakos
2111815fab
The gateway session line printed "◆ Context: 128K tokens (detected)" for a Bedrock model no source knew. _resolve_gateway_model_context recognised only the generic 256K DEFAULT_FALLBACK_CONTEXT as a default, so Bedrock's static 128K was reported as detected. The CLI surfaces showed the number with no source at all. agent/context_pin.py already labels a model.context_length pin "(pinned)" on those surfaces (#66168). This commit extends the same helper with a "fallback" marker: - is_fallback_context_length(window, model=, provider=, base_url=) is True only when the window equals the resolver's default for that route and no local source knows the model. On Bedrock the default is BEDROCK_DEFAULT_CONTEXT_LENGTH, and the checks are the table/family step, a provider-confirmed cache row and model_overrides. The generic route uses DEFAULT_FALLBACK_CONTEXT, the catalog, models.dev and custom_providers. It does only local reads, memoised for 60s because the status bar repaints often. It fails open and never probes. - Labelled surfaces: the gateway session line (via context_source "default", now worded "fallback, not a detected limit"), CLI and gateway /usage, the wide status bar (new cli.status_bar.ctx_fallback key in all 17 locales), the welcome banner, and the CLI and gateway /model switch summary. No new config key. The override is unchanged: model.context_length, or model_overrides.<provider>.<model>.context_window.
1134 lines
59 KiB
Python
1134 lines
59 KiB
Python
"""Model picker, /model switch application, runtime snapshot/restore, and codex runtime handling for the interactive CLI
|
|
|
|
Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. cli.py-internal
|
|
symbols are imported LAZILY inside each method (``from cli import ...``) — the mixin
|
|
never imports ``cli`` at module load time (import cycle).
|
|
|
|
Tests drive the switch paths with bare stubs lacking most mixin methods, so shared steps are
|
|
module-level functions taking ``cli`` and siblings are called as ``HermesCLI.<name>(self, ...)``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import logging
|
|
import sys
|
|
import threading
|
|
|
|
from rich.markup import escape as _escape
|
|
from agent.i18n import t
|
|
from utils import base_url_host_matches
|
|
from hermes_cli.cli_agent_setup_mixin import _retire_agent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# CLI-level fields describing the active model route; snapshotted before a switch / one-turn
|
|
# override and restored wholesale on rollback. ``reasoning_config`` rides along because it is
|
|
# resolved per model: a failed swap or `/model X --reasoning high --once` must not leave the
|
|
# new model's effort behind with the old route.
|
|
_RUNTIME_FIELDS = (
|
|
"model", "provider", "requested_provider", "_explicit_api_key", "_explicit_base_url",
|
|
"api_key", "base_url", "api_mode", "reasoning_config")
|
|
|
|
|
|
def _runtime_fields(cli) -> dict:
|
|
return {key: getattr(cli, key, None) for key in _RUNTIME_FIELDS}
|
|
|
|
|
|
def _resolve_cli_reasoning(cli) -> None:
|
|
"""Re-resolve the CLI-level ``reasoning_config`` for ``cli.model`` through the shared chokepoint
|
|
(per-model ``reasoning_overrides`` > global ``agent.reasoning_effort``). Startup resolves it once
|
|
for the launch model; every path that moves ``cli.model`` must call this BEFORE the agent branch,
|
|
because a lazily built agent inherits this field — an always-thinking model then goes out with
|
|
the launch model's effort and 400s (#112921, #96012). ``agent.switch_model`` re-resolves its own
|
|
copy for the live-agent path."""
|
|
from cli import CLI_CONFIG
|
|
from hermes_constants import resolve_reasoning_config
|
|
# getattr: tests drive /new unbound on a SimpleNamespace without ``model`` (blank -> config default).
|
|
cli.reasoning_config = resolve_reasoning_config(CLI_CONFIG, getattr(cli, "model", None) or "")
|
|
cli._reasoning_user_set = False
|
|
|
|
|
|
def stored_session_route(session_meta, *, current_model, current_provider):
|
|
"""The route a resumed session should run on, or ``None`` when the stored one is absent or
|
|
already current. Returns ``(model, provider, base_url, api_mode, provider_changed)``; the
|
|
canonical row reader is ``SessionDB.session_gateway_runtime`` (``model_config.gateway_runtime``,
|
|
else the TUI's top-level keys). Bare ``custom`` is healed because the CLI resolve path
|
|
hard-fails on it (the TUI gateway keeps it when a base_url exists)."""
|
|
stored_model = str((session_meta or {}).get("model") or "").strip()
|
|
if not stored_model:
|
|
return None
|
|
from hermes_state import SessionDB as _SessionDB
|
|
runtime = _SessionDB.session_gateway_runtime(session_meta)
|
|
base_url = runtime.get("base_url") or None
|
|
api_mode = runtime.get("api_mode") or None
|
|
stored_provider = runtime.get("provider") or None
|
|
provider = _heal_bare_custom_provider(stored_provider, base_url=base_url, model=stored_model)
|
|
if provider != stored_provider:
|
|
# A healed entry resolves its own current endpoint. A route that didn't heal is gone with its
|
|
# endpoint and wire, which must not ride on the ambient provider.
|
|
base_url = None
|
|
if provider is None:
|
|
api_mode = None
|
|
provider_changed = bool(provider) and provider != current_provider
|
|
if stored_model == current_model and not provider_changed:
|
|
return None
|
|
from hermes_cli.runtime_provider import is_foreign_provider_endpoint
|
|
if is_foreign_provider_endpoint(provider, base_url):
|
|
# The endpoint and its wire belong to the provider this chat left; resolve the stored one's own.
|
|
base_url = api_mode = None
|
|
# A row's api_mode/base_url were written for whichever model the session last ran. Providers that
|
|
# pick the wire per model (OpenCode Zen/Go, Nous, plugin profiles) re-derive both from the stored model, or a
|
|
# resumed opencode-go session keeps a MiniMax-era anthropic_messages route for a chat_completions
|
|
# model (#96066) — the CLI/oneshot twin of tui_gateway's _rederive_per_model_route.
|
|
from hermes_cli.model_switch import model_derived_api_mode
|
|
derived = model_derived_api_mode(provider or "", stored_model)
|
|
if derived is not None:
|
|
from hermes_cli.models import normalize_opencode_base_url
|
|
api_mode = derived
|
|
base_url = normalize_opencode_base_url(provider, api_mode, base_url) or None
|
|
return stored_model, provider, base_url, api_mode, provider_changed
|
|
|
|
|
|
def _heal_bare_custom_provider(provider, *, base_url, model):
|
|
"""Bare ``custom`` is a billing class, not a routable identity: persisting/restoring it makes a
|
|
later resume hard-fail once the config default leaves the custom endpoint. Recover the durable
|
|
``custom:<name>`` menu key from the endpoint, else drop the provider (None)."""
|
|
if str(provider or "").strip().lower() != "custom":
|
|
return provider
|
|
try:
|
|
# Heal bare "custom" persisted by older builds / gateway turns: it's the resolved billing class, not
|
|
# a routable identity. (Stricter than the TUI gateway's recovery, which keeps bare "custom" when a
|
|
# base_url exists — the CLI's resolve path would hard-fail on it, #14676.)
|
|
from hermes_cli.runtime_provider import canonical_custom_identity
|
|
return canonical_custom_identity(base_url=base_url or None, model=model or None) or None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _merge_preflight_warning(cli, result, custom_providers) -> None:
|
|
"""Fold the context-compression preflight warning into ``result`` (fail-soft)."""
|
|
from cli import logger
|
|
if cli.agent is None:
|
|
return
|
|
try:
|
|
from hermes_cli.context_switch_guard import merge_preflight_compression_warning
|
|
# Prefer the fresh inventory list (same source as switch_model / TUI); fall back
|
|
# to the agent-init snapshot.
|
|
merge_preflight_compression_warning(
|
|
result,
|
|
agent=cli.agent,
|
|
messages=list(cli.conversation_history or []),
|
|
custom_providers=custom_providers if custom_providers is not None
|
|
else getattr(cli.agent, "_custom_providers", None),
|
|
config_context_length=getattr(cli.agent, "_config_context_length", None))
|
|
except Exception as exc:
|
|
logger.debug("preflight-compression switch warning failed: %s", exc)
|
|
|
|
|
|
def _print_switch_summary(cli, result, old_model, *, one_turn: bool, strict_context: bool) -> None:
|
|
"""Record the next-turn switch note and print the "Model switched" block.
|
|
|
|
The note is prepended to the next user message (a mid-history system message would break
|
|
providers and prompt caching). ``strict_context``: the typed /model path lets
|
|
context-resolution errors propagate; the picker path swallows them.
|
|
"""
|
|
from cli import _cprint
|
|
from hermes_cli.model_switch import format_model_for_display, resolve_display_context_length
|
|
_display_old = format_model_for_display(old_model)
|
|
_display_new = format_model_for_display(result.new_model)
|
|
cli._pending_model_switch_note = (
|
|
f"[Note: model was just switched from {_display_old} to {_display_new} "
|
|
f"via {result.provider_label or result.target_provider}. "
|
|
f"{'This override applies to the next turn only. ' if one_turn else ''}"
|
|
f"Adjust your self-identification accordingly.]")
|
|
_cprint(f" {t('cli.model.switched', model=_display_new)}")
|
|
_cprint(f" {t('gateway.model.provider_label', provider=result.provider_label or result.target_provider)}")
|
|
if result.target_provider == "moa":
|
|
# The preset name hides who pays: the aggregator runs every tool-loop step (#112359).
|
|
from hermes_cli.moa_config import normalize_moa_config
|
|
moa_cfg = cli.config.get("moa") if isinstance(cli.config, dict) else {}
|
|
agg = normalize_moa_config(moa_cfg)["presets"].get(result.new_model, {}).get("aggregator") or {}
|
|
if agg:
|
|
_acting = f"{agg.get('provider')}:{agg.get('model')}"
|
|
_cprint(f" {t('cli.model.acting_model', model=_acting)}")
|
|
|
|
# Provider-aware context chain: Codex OAuth / Nous / profile live-catalog caps win over the raw
|
|
# models.dev entry (gpt-5.5 is 1.05M on openai but 272K on Codex OAuth).
|
|
mi = result.model_info
|
|
agent = cli.agent
|
|
try:
|
|
ctx = resolve_display_context_length(
|
|
result.new_model, result.target_provider,
|
|
base_url=result.base_url or cli.base_url or "",
|
|
api_key=result.api_key or cli.api_key or "", model_info=mi,
|
|
config_context_length=getattr(agent, "_config_context_length", None) if agent else None,
|
|
custom_providers=getattr(agent, "_custom_providers", None) if agent else None)
|
|
except Exception:
|
|
if strict_context:
|
|
raise
|
|
ctx = None
|
|
if ctx:
|
|
from agent.context_pin import context_window_suffix
|
|
_cprint(f" {t('gateway.model.context_label', tokens=f'{ctx:,}')}" + context_window_suffix(
|
|
ctx, getattr(agent, '_config_context_length', None) if agent else None, model=result.new_model,
|
|
provider=result.target_provider, base_url=result.base_url or cli.base_url or ""))
|
|
if mi:
|
|
if mi.max_output:
|
|
_cprint(f" {t('gateway.model.max_output_label', tokens=f'{mi.max_output:,}')}")
|
|
_cprint(f" {t('gateway.model.capabilities_label', capabilities=mi.format_capabilities())}")
|
|
cache_enabled = (
|
|
(base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower())
|
|
or result.api_mode == "anthropic_messages")
|
|
if cache_enabled:
|
|
_cprint(f" {t('gateway.model.prompt_caching_enabled')}")
|
|
if result.warning_message:
|
|
_cprint(f" ⚠ {result.warning_message}")
|
|
if getattr(result, "usage_notice", ""):
|
|
_cprint(f" ⚠ {result.usage_notice}")
|
|
|
|
|
|
def _switch_model_from(
|
|
cli, raw_input, *, is_global, explicit_provider, user_providers, custom_providers):
|
|
"""``switch_model`` seeded with this CLI's live route."""
|
|
from hermes_cli.model_switch import switch_model
|
|
return switch_model(
|
|
raw_input=raw_input, current_provider=cli.provider or "", current_model=cli.model or "",
|
|
current_base_url=cli.base_url or "", current_api_key=cli.api_key or "", is_global=is_global,
|
|
explicit_provider=explicit_provider, user_providers=user_providers,
|
|
custom_providers=custom_providers)
|
|
|
|
|
|
def _run_confirm_and_apply(cli, target, *args) -> None:
|
|
"""Run a confirm+apply sequence off the UI thread when the TUI is live.
|
|
|
|
The expensive-model modal blocks its thread on a response queue (_prompt_text_input_modal);
|
|
on the prompt_toolkit main thread that freezes rendering, so the modal never appears and the
|
|
switch silently cancels after the 120s timeout.
|
|
"""
|
|
if getattr(cli, "_app", None):
|
|
threading.Thread(target=target, args=args, daemon=True).start()
|
|
else:
|
|
target(*args)
|
|
|
|
|
|
def _one_turn_selection_context(cli):
|
|
"""Context the owner's large-context guard would see, for a ``/model --once`` that does not mutate.
|
|
|
|
A live compressor measurement wins. Otherwise refresh ``session.stats``, then the session row's
|
|
persisted usage anchor (a resume, before this process has run a turn).
|
|
"""
|
|
from hermes_cli.model_selection_guards import SelectionContext, selection_context_for_agent
|
|
agent = getattr(cli, "agent", None)
|
|
if agent is not None and hasattr(agent, "_refresh_stats"):
|
|
agent._refresh_stats()
|
|
live = selection_context_for_agent(agent)
|
|
if live is not None:
|
|
return live
|
|
from agent.usage_anchor import persisted_anchor_tokens
|
|
db = getattr(cli, "_session_db", None)
|
|
sid = getattr(cli, "session_id", None)
|
|
if db is None or not sid or not callable(getattr(db, "get_messages_as_conversation", None)):
|
|
return None
|
|
try:
|
|
messages = db.get_messages_as_conversation(sid)
|
|
except Exception:
|
|
logger.debug("one-turn selection context unavailable", exc_info=True)
|
|
return None
|
|
tokens = persisted_anchor_tokens(db, sid, messages)
|
|
if not tokens:
|
|
return None
|
|
return SelectionContext(context_tokens=tokens, current_model=getattr(cli, "model", None) or None)
|
|
|
|
|
|
def _ask_switch_anyway(cli, title: str, detail: str) -> bool:
|
|
"""The model-guard prompt (switch anyway once, or cancel): True only on an explicit yes."""
|
|
choices = [
|
|
("once", t("cli.model.choice_switch_anyway"), t("cli.model.desc_switch_anyway")),
|
|
("cancel", t("cli.model.choice_cancel"), t("cli.model.desc_keep_current_model"))]
|
|
raw = cli._prompt_text_input_modal(title=f"!!! {title} !!!", detail=detail, choices=choices, timeout=120)
|
|
return cli._normalize_slash_confirm_choice(raw, choices) == "once"
|
|
|
|
|
|
# `/model` usage block: (invocation form, description key). Forms are command syntax (never
|
|
# translated); descriptions are padded to one column at render time.
|
|
_MODEL_USAGE_ROWS = (
|
|
("/model <name>", "cli.model.usage_switch"),
|
|
("/model <name> --global", "cli.model.usage_switch_global"),
|
|
("/model <name> --once", "cli.model.usage_switch_once"),
|
|
("/model <name> --session", "cli.model.usage_switch_session"),
|
|
("/model <name> --provider <slug>", "cli.model.usage_switch_provider_model"),
|
|
("/model <name> --reasoning <level>", "cli.model.usage_switch_reasoning"),
|
|
("/model --provider <slug>", "cli.model.usage_switch_provider"),
|
|
("/model --refresh", "cli.model.usage_refresh"))
|
|
_MODEL_USAGE_COL = 36
|
|
|
|
|
|
def _configured_default_route() -> tuple[str, str]:
|
|
"""``(model, provider)`` config.yaml names NOW (``/model reset``'s target). A fresh read, not the
|
|
startup snapshot: the dashboard or another surface may have changed the default since launch."""
|
|
from hermes_cli.config import load_config_readonly
|
|
model_cfg = (load_config_readonly() or {}).get("model")
|
|
if not isinstance(model_cfg, dict):
|
|
return (model_cfg.strip() if isinstance(model_cfg, str) else ""), ""
|
|
provider = str(model_cfg.get("provider") or "").strip()
|
|
return (str(model_cfg.get("default") or model_cfg.get("model") or "").strip(),
|
|
"" if provider.lower() == "auto" else provider)
|
|
|
|
|
|
def _picker_reasoning_rows() -> list[tuple[str, str]]:
|
|
"""``(value, label)`` rows for the picker's effort step: the canonical ladder, the off state,
|
|
then a keep-current row (empty value = leave the effort alone)."""
|
|
from hermes_constants import VALID_REASONING_EFFORTS
|
|
rows = [(lvl, lvl) for lvl in VALID_REASONING_EFFORTS]
|
|
rows.append(("none", t("cli.model.effort_none")))
|
|
rows.append(("", t("cli.model.effort_keep_current")))
|
|
return rows
|
|
|
|
|
|
def _picker_offers_reasoning(provider_data: dict, model: str) -> bool:
|
|
"""False only when the inventory's capability map says the picked model has no reasoning
|
|
control; unknown capabilities keep the step (a no-op dial beats hiding a real one)."""
|
|
caps = (provider_data or {}).get("capabilities")
|
|
entry = caps.get(model) if isinstance(caps, dict) else None
|
|
return not (isinstance(entry, dict) and entry.get("reasoning") is False)
|
|
|
|
|
|
def _apply_reasoning_after_switch(cli, effort: str, *, persist_global: bool, one_turn: bool = False) -> None:
|
|
"""Apply a ``--reasoning <level>`` that rode along with a model pick. Runs AFTER the swap: the
|
|
agent's ``switch_model`` re-resolves ``reasoning_config`` from config.yaml, so an earlier write
|
|
would be clobbered. Session-scoped unless the pick itself persists (``--global``).
|
|
|
|
``one_turn`` (``--once``): the level rides the one-turn admission (``model_override.reasoning``)
|
|
and the restore snapshot puts the CLI's own level back. It is never a session pin."""
|
|
from cli import CLI_CONFIG, _cprint, _parse_reasoning_config, save_config_value
|
|
parsed = _parse_reasoning_config(effort)
|
|
if parsed is None:
|
|
return
|
|
cli.reasoning_config = parsed
|
|
if one_turn:
|
|
from hermes_cli.gateway_agent import GatewayAgent
|
|
pending = getattr(cli, "_pending_one_turn_model_restore", None)
|
|
if isinstance(pending, dict):
|
|
pending["one_turn_reasoning"] = effort # the agent chat rebuilds carries it too
|
|
if isinstance(cli.agent, GatewayAgent):
|
|
cli.agent.stage_turn_reasoning(parsed)
|
|
elif cli.agent is not None:
|
|
cli.agent.reasoning_config = parsed
|
|
_cprint(f" {t('cli.model.reasoning_effort', effort=effort)}")
|
|
return
|
|
cli._reasoning_user_set = True
|
|
if cli.agent is not None:
|
|
# The gateway session's agent commits an explicit level even when it equals the inherited one.
|
|
pin = getattr(cli.agent, "pin_reasoning", None)
|
|
if pin is not None:
|
|
pin(parsed)
|
|
else:
|
|
cli.agent.reasoning_config = parsed
|
|
saved = persist_global and save_config_value("agent.reasoning_effort", effort)
|
|
if saved:
|
|
CLI_CONFIG.setdefault("agent", {})["reasoning_effort"] = effort
|
|
_cprint(f" {t('cli.model.reasoning_effort_saved' if saved else 'cli.model.reasoning_effort', effort=effort)}")
|
|
|
|
|
|
def _commit_model_switch(
|
|
cli, result, *, persist_global: bool, one_turn: bool = False, picker: bool = False,
|
|
reasoning_effort: str = "") -> None:
|
|
"""Stage + swap, print the summary, persist (session row unless --once; config on --global).
|
|
``picker``: tolerate context-resolution errors and label the config write "(--global)"; the
|
|
typed path additionally records the one-turn restore snapshot. ``reasoning_effort`` (from
|
|
``--reasoning`` or the picker's effort step) is applied after the swap."""
|
|
from cli import HermesCLI, _cprint
|
|
old_model, old_provider = cli.model, getattr(cli, "provider", None)
|
|
# A second ``--once`` before the turn restores to the route before the first one.
|
|
snapshot = (getattr(cli, "_pending_one_turn_model_restore", None)
|
|
or cli._snapshot_model_runtime()) if one_turn else None
|
|
if snapshot is not None:
|
|
# The level belongs to this --once pick only: a second pick without --reasoning drops it.
|
|
# So does the alias it resolved through.
|
|
snapshot = {key: value for key, value in snapshot.items()
|
|
if key not in ("one_turn_reasoning", "one_turn_alias")}
|
|
# ``--once`` must not session.mutate.
|
|
cli._model_switch_one_turn = bool(one_turn)
|
|
try:
|
|
staged = cli._stage_and_swap_model(result, old_model)
|
|
finally:
|
|
cli._model_switch_one_turn = False
|
|
if not staged:
|
|
return
|
|
# A TUI slash worker replays /model on its shadow CLI; the tui_gateway mirror counts the real switch.
|
|
if not getattr(cli, "is_slash_worker", False):
|
|
from hermes_cli.observability.shared_metrics_events import record_model_switch
|
|
record_model_switch(
|
|
from_provider=old_provider, to_provider=result.target_provider, surface="cli", from_model=old_model,
|
|
session_id=getattr(getattr(cli, "agent", None), "session_id", None) or getattr(cli, "session_id", None))
|
|
if not picker:
|
|
cli._pending_one_turn_model_restore = snapshot
|
|
if snapshot is not None and getattr(result, "resolved_via_alias", ""):
|
|
snapshot["one_turn_alias"] = result.resolved_via_alias # the agent chat rebuilds carries it
|
|
_print_switch_summary(cli, result, old_model, one_turn=one_turn, strict_context=not picker)
|
|
if reasoning_effort:
|
|
_apply_reasoning_after_switch(cli, reasoning_effort, persist_global=persist_global and not one_turn,
|
|
one_turn=one_turn)
|
|
if persist_global:
|
|
from hermes_cli.model_switch import persist_model_selection
|
|
persist_model_selection(result)
|
|
_cprint(f" {t('cli.model.saved_global' if picker else 'cli.model.saved')}")
|
|
elif one_turn:
|
|
_cprint(f" {t('cli.model.next_turn_only')}")
|
|
else:
|
|
_cprint(f" {t('cli.model.session_only')}")
|
|
# The row records what THIS session runs even on --global (else a later resume restores the
|
|
# stale creation-time model); --once is restored after one turn and never touches the row.
|
|
if not one_turn:
|
|
HermesCLI._persist_model_switch_to_session(cli, result)
|
|
|
|
|
|
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
|
|
from hermes_cli.inventory import build_models_payload
|
|
from hermes_cli.providers import get_label
|
|
try:
|
|
if ctx is None:
|
|
raise RuntimeError("inventory context unavailable")
|
|
providers = build_models_payload(
|
|
ctx, probe_custom_providers=force_refresh,
|
|
probe_current_custom_provider=not force_refresh,
|
|
capabilities=True, # the effort step hides itself on reasoning-free routes
|
|
# Keep providers whose pool is entirely in cooldown visible: limits are
|
|
# per-model for many providers, so another model may work (#103829) —
|
|
# same contract as the gateway picker (#66584) and aux pickers (#66624).
|
|
for_picker=True,
|
|
# Visibility only, not a faster probe: this call live-probes the current custom
|
|
# endpoint, which historically enjoyed the full 5s discovery budget (#103843).
|
|
fast_custom_probe=False,
|
|
)["providers"]
|
|
except Exception:
|
|
providers = []
|
|
if not providers:
|
|
_cprint(f" {t('cli.model.no_authenticated_providers')}")
|
|
_cprint("")
|
|
for form, desc_key in _MODEL_USAGE_ROWS:
|
|
_cprint(f" {form.ljust(_MODEL_USAGE_COL)} {t(desc_key)}")
|
|
return
|
|
cli._open_model_picker(
|
|
providers, cli.model or "unknown", get_label(cli.provider) if cli.provider else "unknown",
|
|
user_provs=ctx.user_providers if ctx is not None else None,
|
|
custom_provs=ctx.custom_providers if ctx is not None else None)
|
|
|
|
|
|
class CLIModelSwitchMixin:
|
|
"""Model picker, /model switch application, runtime snapshot/restore, and codex runtime handling for the interactive CLI"""
|
|
|
|
def _normalize_model_for_provider(self, resolved_provider: str) -> bool:
|
|
"""Normalize provider-specific model IDs and routing."""
|
|
from cli import _split_model_config_default
|
|
current_model = str(self.model or "").strip()
|
|
if isinstance(self.model, dict):
|
|
current_model, _ = _split_model_config_default(self.model)
|
|
changed = False
|
|
|
|
def _adopt(canonical, notice) -> None:
|
|
"""Adopt ``canonical`` when it differs; ``notice(new)`` builds the warning text."""
|
|
nonlocal current_model, changed
|
|
if canonical and canonical != current_model:
|
|
if not self._model_is_default:
|
|
self._console_print(f"[yellow]⚠️ {notice(canonical)}[/]")
|
|
self.model = canonical
|
|
current_model = canonical
|
|
changed = True
|
|
|
|
def _adopt_with_mode(normalize, api_mode_of, notice) -> bool:
|
|
"""Provider families that also own the wire protocol: adopt id, then sync api_mode."""
|
|
nonlocal changed
|
|
try:
|
|
_adopt(normalize(current_model), notice)
|
|
resolved_mode = api_mode_of(current_model)
|
|
if resolved_mode != self.api_mode:
|
|
self.api_mode = resolved_mode
|
|
changed = True
|
|
except Exception:
|
|
pass
|
|
return changed
|
|
|
|
try:
|
|
from hermes_cli.model_normalize import (
|
|
_AGGREGATOR_PROVIDERS, normalize_model_for_provider)
|
|
if resolved_provider not in _AGGREGATOR_PROVIDERS:
|
|
_adopt(
|
|
normalize_model_for_provider(current_model, resolved_provider),
|
|
lambda new: t("cli.model.normalized_model", old=current_model, new=new, provider=resolved_provider))
|
|
except Exception:
|
|
pass
|
|
|
|
from providers import get_provider_profile
|
|
profile = get_provider_profile(resolved_provider)
|
|
try:
|
|
native_id = profile.normalize_model_id(current_model, api_key=self.api_key or "") if profile else None
|
|
except Exception: # a plugin hook failing must not block startup: generic repair continues
|
|
logger.debug("normalize_model_id(%s) failed", resolved_provider, exc_info=True)
|
|
native_id = None
|
|
if native_id is not None: # a profile-owned id space also re-derives the wire for the new id
|
|
from hermes_cli.model_switch import model_derived_api_mode
|
|
return _adopt_with_mode(
|
|
lambda m: native_id,
|
|
lambda m: model_derived_api_mode(resolved_provider, m, self.api_key or "") or self.api_mode,
|
|
lambda new: t("cli.model.normalized_model", old=current_model, new=new, provider=resolved_provider))
|
|
|
|
from hermes_cli.models import opencode_provider_family
|
|
if opencode_provider_family(resolved_provider) is not None:
|
|
from hermes_cli.models import normalize_opencode_model_id, opencode_model_api_mode
|
|
return _adopt_with_mode(
|
|
lambda m: normalize_opencode_model_id(resolved_provider, m),
|
|
lambda m: opencode_model_api_mode(resolved_provider, m),
|
|
lambda new: t("cli.model.stripped_provider_prefix", old=current_model, new=new, provider=resolved_provider))
|
|
|
|
if resolved_provider != "openai-codex":
|
|
return changed
|
|
|
|
# 1. Strip provider prefix ("openai/gpt-5.4" → "gpt-5.4")
|
|
if "/" in current_model:
|
|
slug = current_model.split("/", 1)[1]
|
|
if not self._model_is_default:
|
|
self._console_print(
|
|
"[yellow]⚠️ " + _escape(t("cli.model.stripped_provider_prefix", old=current_model, new=slug, provider="OpenAI Codex")) + "[/]")
|
|
self.model = slug
|
|
current_model = slug
|
|
changed = True
|
|
|
|
# 2. Replace untouched default with a Codex model
|
|
if self._model_is_default:
|
|
from hermes_cli.codex_models import DEFAULT_CODEX_MODELS, get_codex_model_ids
|
|
|
|
fallback_model = DEFAULT_CODEX_MODELS[0]
|
|
try:
|
|
# self.base_url is the route resolved with self.api_key (#121486).
|
|
available = get_codex_model_ids(
|
|
access_token=self.api_key if self.api_key else None, base_url=self.base_url or None)
|
|
if available:
|
|
fallback_model = available[0]
|
|
except Exception:
|
|
pass
|
|
if current_model != fallback_model:
|
|
self.model = fallback_model
|
|
changed = True
|
|
return changed
|
|
|
|
def _persist_model_switch_to_session(self, result) -> None:
|
|
"""Persist a session-scoped /model switch to the session DB row.
|
|
|
|
Writes the model column plus the route in both shapes readers use — nested
|
|
``gateway_runtime`` (CLI --resume) and top-level keys (TUI session.resume) — from one
|
|
or-None dict so stale keys are DELETED (``_merge_model_config_json`` only deletes on
|
|
explicit None) and the shapes never diverge.
|
|
|
|
Writes the model column plus the runtime route so ``--resume`` (CLI, reads ``gateway_runtime``) and
|
|
``session.resume`` (TUI/desktop, reads top-level ``model_config`` keys via
|
|
``_stored_session_runtime_overrides``) both restore the switched provider instead of recombining the
|
|
model with the ambient default (#79536). Mirrors the gateway's ``update_session_model()`` call.
|
|
getattr: tests drive the switch paths with ``object.__new__`` stubs.
|
|
"""
|
|
from cli import logger
|
|
db = getattr(self, "_session_db", None)
|
|
sid = getattr(self, "session_id", None)
|
|
if not db or not sid:
|
|
return
|
|
route = {
|
|
"provider": _heal_bare_custom_provider(
|
|
result.target_provider, base_url=result.base_url, model=result.new_model,
|
|
) or None,
|
|
# Both shapes use the same or-None discipline so stale keys from a previous switch are deleted
|
|
# (not merely omitted) in BOTH the nested gateway_runtime dict (CLI reader) and the top-level
|
|
# keys (TUI gateway reader). _merge_model_config_json only deletes on explicit None, so falsy
|
|
# values must be converted, not filtered. Deriving the top-level from **route guarantees the two
|
|
# shapes can never diverge — the asymmetry that caused the original stale-key bug (#85261
|
|
# simplify-code review).
|
|
"base_url": result.base_url or None,
|
|
"api_mode": result.api_mode or None}
|
|
try:
|
|
db.update_session_model(sid, result.new_model)
|
|
db.patch_session_model_config(sid, {"gateway_runtime": route, **route})
|
|
except Exception:
|
|
logger.debug("Failed to persist model switch to session DB", exc_info=True)
|
|
|
|
def _restore_session_model(self, session_meta: dict, *, quiet: bool = False) -> None:
|
|
"""Restore model/provider from the session DB row on every resume path.
|
|
|
|
Skips when no model is recorded or the CLI got an explicit ``-m`` (user intent wins).
|
|
A different stored provider gets its credentials re-resolved — the ambient ``api_key``
|
|
must not be sent to the session's endpoint; on failure the ambient credentials are kept
|
|
so the session still opens (the first turn surfaces the auth error).
|
|
"""
|
|
from cli import logger
|
|
if not (session_meta or {}).get("model") or getattr(self, "_explicit_model_override", False):
|
|
return
|
|
route = stored_session_route(session_meta, current_model=self.model, current_provider=self.provider)
|
|
if route is None:
|
|
return
|
|
stored_model, stored_provider, stored_base_url, stored_api_mode, provider_changed = route
|
|
from hermes_cli.local_runtime.endpoint import LLAMACPP_ALIASES
|
|
managed = str(stored_provider or "").strip().lower() in LLAMACPP_ALIASES
|
|
self.model = stored_model
|
|
if stored_provider:
|
|
self.provider = stored_provider
|
|
self.requested_provider = stored_provider
|
|
if stored_base_url and not managed:
|
|
self.base_url = stored_base_url
|
|
if stored_api_mode:
|
|
self.api_mode = stored_api_mode
|
|
if managed and not (getattr(self, "_explicit_base_url", None) and not provider_changed):
|
|
# The supervisor owns the live port: last boot's loopback URL (an ephemeral fallback when
|
|
# 18434 was busy) must not pin the resume onto a dead endpoint. A launch-time --base-url
|
|
# for this same provider is user intent and keeps winning.
|
|
self._explicit_api_key = None
|
|
self._explicit_base_url = None
|
|
try:
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider
|
|
resolved = resolve_runtime_provider(requested=stored_provider, target_model=self.model or None)
|
|
if resolved.get("api_key"):
|
|
self.api_key = resolved["api_key"]
|
|
self._credential_pool = resolved.get("credential_pool")
|
|
if resolved.get("base_url"):
|
|
self.base_url = resolved["base_url"]
|
|
if not stored_api_mode and resolved.get("api_mode"):
|
|
self.api_mode = resolved["api_mode"]
|
|
except Exception:
|
|
if stored_base_url:
|
|
self.base_url = stored_base_url
|
|
logger.debug(
|
|
"Credential re-resolution for resumed session provider "
|
|
"%s failed; keeping ambient credentials",
|
|
stored_provider, exc_info=True)
|
|
elif provider_changed:
|
|
# Launch-time explicit overrides belong to the AMBIENT provider and would poison
|
|
# _ensure_runtime_credentials for the restored one. api_key is never persisted to
|
|
# the session DB — runtime provider resolution owns credentials.
|
|
self._explicit_api_key = None
|
|
self._explicit_base_url = stored_base_url
|
|
try:
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider
|
|
resolved = resolve_runtime_provider(requested=stored_provider, target_model=self.model or None)
|
|
if resolved.get("api_key"):
|
|
self.api_key = resolved["api_key"]
|
|
self._credential_pool = resolved.get("credential_pool")
|
|
if not stored_base_url and resolved.get("base_url"):
|
|
self.base_url = resolved["base_url"]
|
|
if not stored_api_mode and resolved.get("api_mode"):
|
|
self.api_mode = resolved["api_mode"]
|
|
except Exception:
|
|
logger.debug(
|
|
"Credential re-resolution for resumed session provider "
|
|
"%s failed; keeping ambient credentials",
|
|
stored_provider, exc_info=True)
|
|
_resolve_cli_reasoning(self)
|
|
# Mid-chat /resume swaps the live agent; on startup --resume _init_agent picks up
|
|
# self.model / self.provider / self.reasoning_config.
|
|
if self.agent is not None:
|
|
try:
|
|
self.agent.switch_model(
|
|
new_model=self.model, new_provider=self.provider, api_key=self.api_key or "",
|
|
base_url=self.base_url or "", api_mode=self.api_mode or "")
|
|
except Exception:
|
|
logger.debug("In-place agent model swap on resume failed", exc_info=True)
|
|
msg = t("cli.model.restored_from_session", model=stored_model)
|
|
if stored_provider:
|
|
msg += f" ({stored_provider})"
|
|
if quiet:
|
|
print(msg, file=sys.stderr)
|
|
else:
|
|
self._console_print(f"[dim]{_escape(msg)}[/dim]")
|
|
|
|
def _open_model_picker(self, providers: list, current_model: str, current_provider: str, user_provs=None, custom_provs=None) -> None:
|
|
"""Open prompt_toolkit-native /model picker modal."""
|
|
self._capture_modal_input_snapshot()
|
|
self._model_picker_state = {
|
|
"stage": "provider",
|
|
"providers": providers,
|
|
"selected": next((i for i, p in enumerate(providers) if p.get("is_current")), 0),
|
|
"current_model": current_model,
|
|
"current_provider": current_provider,
|
|
"user_provs": user_provs,
|
|
"custom_provs": custom_provs,
|
|
"filter": ""}
|
|
self._invalidate(min_interval=0.0)
|
|
|
|
def _confirm_expensive_model_switch(self, result) -> bool:
|
|
"""Ask for explicit confirmation before applying costly model switches."""
|
|
self._confirmed_switch_warning = None
|
|
if not getattr(result, "success", False):
|
|
return True
|
|
try:
|
|
from hermes_cli.model_selection_guards import (
|
|
combined_selection_warning, selection_context_for_agent)
|
|
primed = getattr(self, "_one_turn_selection_context", None)
|
|
warning = combined_selection_warning(
|
|
result.new_model, provider=result.target_provider,
|
|
base_url=result.base_url or self.base_url or "",
|
|
api_key=result.api_key or self.api_key or "", model_info=result.model_info,
|
|
selection_context=primed if primed is not None else selection_context_for_agent(
|
|
getattr(self, "agent", None)))
|
|
except Exception:
|
|
warning = None
|
|
if warning is None:
|
|
return True
|
|
if not _ask_switch_anyway(self, warning.title, warning.message):
|
|
return False
|
|
self._confirmed_switch_warning = warning.message
|
|
return True
|
|
|
|
def _confirm_owner_model_switch(self, refusal) -> bool:
|
|
"""The gateway owner refused a guarded target (``confirmation_required``: cost, data policy,
|
|
large context; Ink/Desktop/ACP show the same ``confirm_message``): ask in the same prompt as
|
|
``_confirm_expensive_model_switch``. Only a yes re-sends with the owner's token. The identical
|
|
warning the user just confirmed for this switch is not asked twice."""
|
|
from hermes_cli.gateway_mutations import confirmation_title
|
|
message = refusal.get("confirm_message") or ""
|
|
confirmed, self._confirmed_switch_warning = getattr(self, "_confirmed_switch_warning", None), None
|
|
if message and message == confirmed:
|
|
return True
|
|
return _ask_switch_anyway(self, confirmation_title(refusal), message)
|
|
|
|
def _confirm_and_apply_model_switch_result(
|
|
self, result, persist_global: bool, custom_providers=None, reasoning_effort: str = "") -> None:
|
|
from cli import _cprint
|
|
try:
|
|
if result.success and not self._confirm_expensive_model_switch(result):
|
|
_cprint(f" {t('cli.model.switch_cancelled')}")
|
|
return
|
|
self._apply_model_switch_result(
|
|
result, persist_global, custom_providers=custom_providers, reasoning_effort=reasoning_effort)
|
|
except Exception as exc:
|
|
_cprint(f" {t('cli.model.selection_failed', error=exc)}")
|
|
|
|
def _close_model_picker(self) -> None:
|
|
self._model_picker_state = None
|
|
self._restore_modal_input_snapshot()
|
|
self._invalidate(min_interval=0.0)
|
|
|
|
def _snapshot_model_runtime(self) -> dict:
|
|
"""Capture current CLI and agent model runtime for one-turn restore."""
|
|
agent = getattr(self, "agent", None)
|
|
# ``reasoning_config`` is a mutable dict: deepcopy it so a later in-place edit cannot alias the snapshot.
|
|
return {
|
|
**_runtime_fields(self),
|
|
"reasoning_config": copy.deepcopy(getattr(self, "reasoning_config", None)),
|
|
"agent_primary_runtime": copy.deepcopy(
|
|
getattr(agent, "_primary_runtime", None)
|
|
) if agent is not None else None}
|
|
|
|
def _restore_model_runtime_snapshot(self, snapshot: dict | None) -> None:
|
|
"""Restore a model runtime captured before a one-turn override."""
|
|
from cli import _cprint, logger
|
|
if not snapshot:
|
|
return
|
|
one_turn = _runtime_fields(self)
|
|
for key in _RUNTIME_FIELDS:
|
|
if key in snapshot:
|
|
setattr(self, key, snapshot.get(key))
|
|
|
|
agent = getattr(self, "agent", None)
|
|
if agent is None:
|
|
return
|
|
from hermes_cli.gateway_agent import GatewayAgent
|
|
# A gateway agent's one-turn level only rode the admission: put it back locally, write nothing.
|
|
restore_reasoning = (agent.restore_reasoning if isinstance(agent, GatewayAgent)
|
|
else lambda config: setattr(agent, "reasoning_config", config))
|
|
if "reasoning_config" in snapshot:
|
|
restore_reasoning(snapshot["reasoning_config"])
|
|
primary = snapshot.get("agent_primary_runtime")
|
|
if primary and hasattr(agent, "_restore_primary_runtime"):
|
|
try:
|
|
agent._primary_runtime = copy.deepcopy(primary)
|
|
agent._fallback_activated = True
|
|
agent._rate_limited_until = 0
|
|
if agent._restore_primary_runtime():
|
|
return
|
|
except Exception:
|
|
logger.debug("CLI one-turn model restore via primary runtime failed", exc_info=True)
|
|
if hasattr(agent, "switch_model"):
|
|
try:
|
|
# The classic CLI only builds a GatewayAgent (``_init_agent``): its ``once`` / ``restore`` are
|
|
# named, so a rename fails loudly instead of turning the restore into a session mutate.
|
|
restored = agent.switch_model(
|
|
new_model=snapshot.get("model", ""), new_provider=snapshot.get("provider", ""),
|
|
api_key=snapshot.get("api_key", ""), base_url=snapshot.get("base_url", ""),
|
|
api_mode=snapshot.get("api_mode", ""),
|
|
capabilities=snapshot.get("capabilities"), once=True, restore=True)
|
|
except Exception as exc: # health: allow BLE001 -- runs in the turn's finally; reported below
|
|
logger.warning("CLI one-turn model restore failed: %s", exc)
|
|
restored = exc
|
|
if restored is False or isinstance(restored, Exception):
|
|
# The previous route did not come back. Stay on the one-turn model instead of
|
|
# claiming the restore.
|
|
for key, value in one_turn.items():
|
|
setattr(self, key, value)
|
|
notice = t("cli.model.switch_cancelled") if restored is False else t(
|
|
"cli.model.switch_failed", model=snapshot.get("model", ""), error=restored, old_model=agent.model)
|
|
_cprint(f" {notice}")
|
|
return
|
|
if "reasoning_config" in snapshot:
|
|
restore_reasoning(snapshot["reasoning_config"])
|
|
|
|
@staticmethod
|
|
def _filter_model_picker_entries(entries: list, query: str) -> list:
|
|
"""Return (original_index, label) pairs matching ``query`` (case-insensitive subsequence;
|
|
empty matches all). The ORIGINAL index keeps a filtered selection resolving to exactly one
|
|
concrete model — filtering never introduces fuzzy *resolution*."""
|
|
pairs = list(enumerate(entries))
|
|
q = (query or "").strip().lower()
|
|
if not q:
|
|
return pairs
|
|
|
|
def _subseq(needle: str, hay: str) -> bool:
|
|
it = iter(hay)
|
|
return all(ch in it for ch in needle)
|
|
|
|
return [(i, e) for (i, e) in pairs if _subseq(q, str(e).lower())]
|
|
|
|
@staticmethod
|
|
def _compute_model_picker_viewport(
|
|
selected: int, scroll_offset: int, n: int, term_rows: int, reserved_below: int = 6,
|
|
panel_chrome: int = 6, min_visible: int = 3) -> tuple[int, int]:
|
|
"""Resolve (scroll_offset, visible) for the /model picker viewport. ``reserved_below``
|
|
matches the approval/clarify panels (input, status bar, separators); ``panel_chrome`` is
|
|
borders + blanks + hint row. The offset slides to keep ``selected`` on screen."""
|
|
max_visible = max(min_visible, term_rows - reserved_below - panel_chrome)
|
|
if n <= max_visible:
|
|
return 0, n
|
|
visible = max_visible
|
|
if selected < scroll_offset:
|
|
scroll_offset = selected
|
|
elif selected >= scroll_offset + visible:
|
|
scroll_offset = selected - visible + 1
|
|
return max(0, min(scroll_offset, n - visible)), visible
|
|
|
|
def _stage_and_swap_model(self, result, old_model) -> bool:
|
|
"""Stage ``result`` onto the CLI fields, then swap the live agent in place.
|
|
|
|
CLI fields are snapshotted first so a failed agent swap rolls the whole CLI back —
|
|
otherwise the staged broken credentials leak into the next turn even though the agent
|
|
rolled back. Returns False after printing the failure (a failed switch is a no-op).
|
|
"""
|
|
from cli import _cprint
|
|
if self.agent is None and getattr(self, "_gateway_session_open", False) and not self._init_agent():
|
|
# Before the first turn: the stand-in built for that turn reads the gateway session's model,
|
|
# so the switch must commit on the session now; _init_agent printed why it could not.
|
|
self._confirmed_switch_warning = None
|
|
return False
|
|
_cli_snapshot = _runtime_fields(self)
|
|
self.model = result.new_model
|
|
self.provider = result.target_provider
|
|
self.requested_provider = result.target_provider
|
|
# Always overwrite explicit overrides so stale credentials from the previous provider
|
|
# (e.g. Ollama api_key/base_url) don't leak into the next resolution.
|
|
self._explicit_api_key = result.api_key
|
|
self._explicit_base_url = result.base_url
|
|
if result.api_key:
|
|
self.api_key = result.api_key
|
|
if result.base_url:
|
|
self.base_url = result.base_url
|
|
if result.api_mode:
|
|
self.api_mode = result.api_mode
|
|
_resolve_cli_reasoning(self)
|
|
|
|
if self.agent is not None:
|
|
try:
|
|
fields = dict(
|
|
new_model=result.new_model, new_provider=result.target_provider,
|
|
api_key=result.api_key, base_url=result.base_url, api_mode=result.api_mode,
|
|
capabilities=getattr(result, "runtime_capabilities", None))
|
|
if getattr(self, "_model_switch_one_turn", False):
|
|
# The CLI and agent show the resolved model; the admission sends the alias.
|
|
fields.update(once=True, alias=getattr(result, "resolved_via_alias", "") or "")
|
|
if self.agent.switch_model(**fields) is False:
|
|
# Declined at the gateway owner's guard prompt: nothing was committed.
|
|
for _k, _v in _cli_snapshot.items():
|
|
setattr(self, _k, _v)
|
|
_cprint(f" {t('cli.model.switch_cancelled')}")
|
|
return False
|
|
except Exception as exc:
|
|
# The agent rolled itself back to the old working model/client. Roll the CLI's own staged
|
|
# fields back too and abort the rest of the commit (note + success print) so a failed switch
|
|
# is a no-op rather than a dead session (#50163).
|
|
# Agent rolled itself back; roll the CLI back too and abort so a failed switch is a no-op
|
|
# rather than a dead session (#50163).
|
|
for _k, _v in _cli_snapshot.items():
|
|
setattr(self, _k, _v)
|
|
if str(exc) == "launch_credentials_unavailable":
|
|
# No model switch can fix a chat whose launch key changed; say what does.
|
|
from hermes_cli.cli_chat_error_copy import refusal_text
|
|
_cprint(f" {refusal_text(exc)}")
|
|
return False
|
|
_cprint(f" {t('cli.model.switch_failed', model=result.new_model, error=exc, old_model=old_model)}")
|
|
return False
|
|
finally:
|
|
self._confirmed_switch_warning = None # this switch's confirmation only
|
|
return True
|
|
|
|
def _apply_model_switch_result(
|
|
self, result, persist_global: bool, custom_providers=None, reasoning_effort: str = "") -> None:
|
|
"""Picker-path commit (see _commit_model_switch)."""
|
|
from cli import _cprint
|
|
if not result.success:
|
|
_cprint(f" ✗ {result.error_message}")
|
|
return
|
|
_merge_preflight_warning(self, result, custom_providers)
|
|
_commit_model_switch(self, result, persist_global=persist_global, picker=True,
|
|
reasoning_effort=reasoning_effort)
|
|
|
|
def _handle_model_picker_selection(self, persist_global: bool = False) -> None:
|
|
state = self._model_picker_state
|
|
if not state:
|
|
return
|
|
selected = state.get("selected", 0)
|
|
stage = state.get("stage")
|
|
if stage == "provider":
|
|
providers = state.get("providers") or []
|
|
if selected >= len(providers):
|
|
self._close_model_picker()
|
|
return
|
|
provider_data = providers[selected]
|
|
# Curated list (same as `hermes model` / gateway pickers); live catalog only when
|
|
# it is empty (user-defined endpoints, per-resource providers such as azure-foundry).
|
|
# Disk-cached like the gateway pickers: the live probe can walk several api-version
|
|
# fallbacks with a 6 s timeout each, which must not block the REPL on every select.
|
|
model_list = provider_data.get("models", [])
|
|
if not model_list:
|
|
try:
|
|
from hermes_cli.models import cached_provider_model_ids
|
|
model_list = cached_provider_model_ids(provider_data["slug"]) or model_list
|
|
except Exception:
|
|
pass
|
|
from hermes_cli.models_validate import offered_model_ids
|
|
model_list = offered_model_ids(
|
|
model_list, provider_data.get("slug"), provider_data.get("api_url"))
|
|
state.update(
|
|
stage="model", provider_data=provider_data, model_list=model_list,
|
|
selected=0, filter="", _filtered_pairs=None)
|
|
self._invalidate(min_interval=0.0)
|
|
return
|
|
if stage == "model":
|
|
provider_data = state.get("provider_data") or {}
|
|
model_list = state.get("model_list") or []
|
|
# Map the row through the active fuzzy filter; pairs carry the ORIGINAL index.
|
|
filtered_pairs = state.get("_filtered_pairs")
|
|
if filtered_pairs is None:
|
|
filtered_pairs = list(enumerate(model_list))
|
|
visible_labels = [e for (_i, e) in filtered_pairs]
|
|
back_idx = len(visible_labels)
|
|
if selected == back_idx:
|
|
state.update(
|
|
stage="provider", filter="", _filtered_pairs=None,
|
|
selected=next((i for i, p in enumerate(state.get("providers") or [])
|
|
if p.get("slug") == provider_data.get("slug")), 0))
|
|
self._invalidate(min_interval=0.0)
|
|
return
|
|
if selected > back_idx: # cancel row (and anything past it)
|
|
self._close_model_picker()
|
|
return
|
|
if 0 <= selected < back_idx:
|
|
result = _switch_model_from(
|
|
self, visible_labels[selected], is_global=persist_global,
|
|
explicit_provider=provider_data.get("slug"),
|
|
user_providers=state.get("user_provs"),
|
|
custom_providers=state.get("custom_provs"))
|
|
if result.success and _picker_offers_reasoning(provider_data, result.new_model):
|
|
# Third step: effort for the picked model (skipped for routes the catalog
|
|
# marks reasoning-free). Rows come from the canonical level set.
|
|
state.update(stage="reasoning", switch_result=result, selected=0, _scroll_offset=0)
|
|
self._invalidate(min_interval=0.0)
|
|
return
|
|
self._commit_picker_result(result, persist_global)
|
|
return
|
|
self._close_model_picker()
|
|
if stage == "reasoning":
|
|
rows = _picker_reasoning_rows()
|
|
result = state.get("switch_result")
|
|
if selected == len(rows): # ← Back to the model list
|
|
state.update(stage="model", selected=0, _scroll_offset=0, switch_result=None)
|
|
self._invalidate(min_interval=0.0)
|
|
return
|
|
if selected > len(rows) or result is None:
|
|
self._close_model_picker()
|
|
return
|
|
self._commit_picker_result(result, persist_global, reasoning_effort=rows[selected][0])
|
|
|
|
def _commit_picker_result(self, result, persist_global: bool, reasoning_effort: str = "") -> None:
|
|
"""Close the picker and run the confirm+apply sequence for ``result``."""
|
|
state = self._model_picker_state or {}
|
|
# Capture before close — picker state is cleared on close.
|
|
_picker_custom_provs = state.get("custom_provs")
|
|
self._close_model_picker()
|
|
# The effort is appended only when picked: stubs/tests pin the historical arity.
|
|
extra = (reasoning_effort,) if reasoning_effort else ()
|
|
_run_confirm_and_apply(
|
|
self, self._confirm_and_apply_model_switch_result,
|
|
result, persist_global, _picker_custom_provs, *extra)
|
|
|
|
def _handle_model_switch(self, cmd_original: str):
|
|
"""Handle /model command — switch model.
|
|
|
|
Supports:
|
|
/model — show current model + usage hints
|
|
/model <name> — switch model (this session only)
|
|
/model <name> --once — switch for the next turn only
|
|
/model <name> --session — switch for this session only (explicit)
|
|
/model <name> --global — switch and persist to config.yaml
|
|
/model <name> --provider <provider> — switch provider + model
|
|
/model <name> --reasoning <level> — switch and set reasoning effort in one step
|
|
/model --provider <provider> — switch to provider, auto-detect model
|
|
/model reset — back to config.yaml's model (keeps history)
|
|
|
|
Switches are session-scoped unless ``model.persist_switch_by_default`` or ``--global``.
|
|
"""
|
|
from cli import _cprint
|
|
from hermes_cli.model_switch import parse_model_switch_args, resolve_persist_behavior
|
|
|
|
parts = cmd_original.split(None, 1) # split off '/model'
|
|
request = parse_model_switch_args(parts[1].strip() if len(parts) > 1 else "")
|
|
if request.errors:
|
|
# CLI decoration: " ✗ " prefix over the canonical error copy.
|
|
_cprint(f" ✗ {request.error_messages()[0]}")
|
|
return
|
|
target, explicit_provider = request.target, request.explicit_provider
|
|
if request.is_reset:
|
|
# Session-scoped switch back to config.yaml's model; drops an armed --once restore,
|
|
# which would otherwise re-plant the model being reset away from after the next turn.
|
|
target, explicit_provider = _configured_default_route()
|
|
if not target:
|
|
_cprint(f" ✗ {t('cli.model.reset_no_default')}")
|
|
return
|
|
if target == self.model and explicit_provider in ("", self.provider):
|
|
self._pending_one_turn_model_restore = None
|
|
_cprint(f" {t('cli.model.reset_none', model=target)}")
|
|
return
|
|
one_turn = request.is_once
|
|
persist_global = not request.is_reset and resolve_persist_behavior(
|
|
request.is_global, request.is_session, is_once=one_turn,
|
|
explicit_provider=request.explicit_provider)
|
|
|
|
# --refresh: wipe the picker cache so every authed provider's /v1/models is re-fetched.
|
|
if request.force_refresh:
|
|
try:
|
|
from hermes_cli.models import clear_provider_models_cache
|
|
clear_provider_models_cache()
|
|
_cprint(f" {t('cli.model.cleared_picker_cache')}")
|
|
except Exception:
|
|
pass
|
|
|
|
# Live session state is overlaid truthy-only so empty self.* attrs don't clobber config.
|
|
from hermes_cli.inventory import load_picker_context
|
|
try:
|
|
ctx = load_picker_context().with_overrides(
|
|
current_provider=self.provider or "", current_model=self.model or "",
|
|
current_base_url=self.base_url or "")
|
|
except Exception:
|
|
ctx = None
|
|
# switch_model() + _open_model_picker still need the raw provider dicts.
|
|
user_provs = ctx.user_providers if ctx is not None else None
|
|
custom_provs = ctx.custom_providers if ctx is not None else None
|
|
|
|
if not request.target and not request.explicit_provider:
|
|
return _show_model_picker(self, ctx, request.force_refresh)
|
|
|
|
result = _switch_model_from(
|
|
self, target, is_global=persist_global, explicit_provider=explicit_provider,
|
|
user_providers=user_provs, custom_providers=custom_provs)
|
|
if not result.success:
|
|
_cprint(f" ✗ {result.error_message}")
|
|
return
|
|
_merge_preflight_warning(self, result, custom_provs)
|
|
extra = (request.reasoning_effort,) if request.reasoning_effort else ()
|
|
_run_confirm_and_apply(
|
|
self, self._confirm_and_apply_cli_model_switch,
|
|
result, persist_global, one_turn, custom_provs, *extra)
|
|
|
|
def _confirm_and_apply_cli_model_switch(
|
|
self, result, persist_global: bool, one_turn: bool, custom_provs=None, reasoning_effort: str = "") -> None:
|
|
"""Confirm an expensive model switch and apply it (typed /model path). Runs on a worker
|
|
thread when the TUI is active (see _run_confirm_and_apply) so the modal can render."""
|
|
from cli import _cprint
|
|
# ``--once`` does not reach the owner's mutate guard. A resumed session's compressor is
|
|
# still cold; the persisted anchor is what that guard would have measured.
|
|
self._one_turn_selection_context = _one_turn_selection_context(self) if one_turn else None
|
|
try:
|
|
if not self._confirm_expensive_model_switch(result):
|
|
_cprint(f" {t('cli.model.switch_cancelled')}")
|
|
return
|
|
_commit_model_switch(self, result, persist_global=persist_global, one_turn=one_turn,
|
|
reasoning_effort=reasoning_effort)
|
|
finally:
|
|
self._one_turn_selection_context = None
|
|
|
|
def _handle_codex_runtime(self, cmd_original: str) -> None:
|
|
"""Handle /codex-runtime — toggle the codex app-server runtime opt-in.
|
|
|
|
Usage:
|
|
/codex-runtime — show current state
|
|
/codex-runtime auto — Hermes default (chat_completions)
|
|
/codex-runtime codex_app_server — hand turns to codex subprocess
|
|
/codex-runtime on / off — synonyms for the above
|
|
"""
|
|
from cli import _cprint
|
|
from hermes_cli import codex_runtime_switch as crs
|
|
|
|
parts = cmd_original.split(None, 1)
|
|
new_value, errors = crs.parse_args(parts[1].strip() if len(parts) > 1 else "")
|
|
if errors:
|
|
for err in errors:
|
|
_cprint(f"❌ {err}")
|
|
return
|
|
try:
|
|
from hermes_cli.config import load_config, save_config
|
|
except Exception as exc:
|
|
_cprint(t("cli.model.could_not_load_config", error=exc))
|
|
return
|
|
result = crs.apply(
|
|
load_config(), new_value,
|
|
persist_callback=(save_config if new_value is not None else None))
|
|
prefix = "✓" if result.success else "✗"
|
|
for line in result.message.splitlines():
|
|
_cprint(f" {prefix} {line}" if line.startswith("openai_runtime") else f" {line}")
|
|
if result.success and result.requires_new_session:
|
|
_cprint(f" {t('cli.model.tip_reset_new_session')}")
|
|
|
|
def _should_handle_model_command_inline(self, text: str, has_images: bool = False) -> bool:
|
|
"""Return True when /model should be handled immediately on the UI thread."""
|
|
from cli import _looks_like_slash_command
|
|
if not text or has_images or not _looks_like_slash_command(text):
|
|
return False
|
|
try:
|
|
from hermes_cli.commands import resolve_command
|
|
cmd = resolve_command(text.split(None, 1)[0].lower().lstrip('/'))
|
|
return bool(cmd and cmd.name == "model")
|
|
except Exception:
|
|
return False
|
|
|
|
def _cmd_moa(self, cmd_original: str):
|
|
"""/moa one-shot: run one prompt through the default MoA preset, then restore the prior
|
|
model (a session-long MoA switch goes through the picker's virtual MoA provider)."""
|
|
from cli import _cprint, _slash_args
|
|
from hermes_cli.moa_config import moa_usage, normalize_moa_config
|
|
|
|
payload = _slash_args(cmd_original)
|
|
if not payload:
|
|
_cprint(f" {moa_usage()}")
|
|
return True
|
|
moa_cfg = self.config.get("moa") if isinstance(self.config, dict) else {}
|
|
preset = normalize_moa_config(moa_cfg)["default_preset"]
|
|
self._pending_moa_restore_model = {
|
|
key: getattr(self, key, None)
|
|
for key in (
|
|
"requested_provider", "provider", "model", "api_key", "base_url", "api_mode")}
|
|
self.requested_provider = "moa"
|
|
self.provider = "moa"
|
|
self.model = preset
|
|
self.api_key = "moa-virtual-provider"
|
|
self.base_url = "moa://local"
|
|
self.api_mode = "chat_completions"
|
|
_retire_agent(self)
|
|
self._pending_moa_disable_after_turn = True
|
|
self._pending_agent_seed = payload
|
|
_cprint(f" {t('cli.model.moa_one_shot_queued', preset=preset)}")
|