mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
unsupportedpastels
f19a64fd0a
A persisting /model pick writes persist_model_selection's whole plan, not just model.default/provider/base_url: model.api_mode is always set or cleared, and model.context_length plus stale credential keys are cleared on a route change. A pin on one of those passed the check, so the pick switched the live session, wrote the user's file and reported success while the administrator's value kept winning. The immediate path now checks the exact plan from model_selection_config_updates once switch_model has resolved the target, still before any confirm prompt or live swap. A pick queued mid-turn is resolved only at turn start, so it counts every clear the raw model block makes possible. Without a managed scope the check returns before reading anything.
565 lines
30 KiB
Python
565 lines
30 KiB
Python
"""Model switching for a live session: persist, snapshot/restore runtime, /model apply with
|
|
guards, bot-capability + config sync. Bodies are rebound onto server.py's globals at install
|
|
time (method_ctx.bind_module), so they reference server.py globals bare."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import copy
|
|
|
|
from .method_ctx import HandlerRegistry, bind_module
|
|
|
|
_registry = HandlerRegistry()
|
|
|
|
|
|
_RUNTIME_KEYS = ("model", "provider", "api_key", "base_url", "api_mode")
|
|
|
|
|
|
def _snapshot_agent_model_runtime(agent) -> dict:
|
|
"""Capture the current agent model runtime for a one-turn restore."""
|
|
return {**{k: getattr(agent, k, "") for k in _RUNTIME_KEYS},
|
|
"reasoning_config": copy.deepcopy(getattr(agent, "reasoning_config", None)),
|
|
"primary_runtime": copy.deepcopy(getattr(agent, "_primary_runtime", None))}
|
|
|
|
|
|
def _restore_agent_model_runtime(agent, snapshot: dict | None) -> None:
|
|
"""Restore an agent model runtime captured before a one-turn override."""
|
|
if not snapshot or agent is None:
|
|
return
|
|
# `/model X --reasoning high --once`: the effort leaves with the model. Set before the
|
|
# runtime restore paths below (primary_runtime may predate a session /reasoning change).
|
|
if "reasoning_config" in snapshot:
|
|
agent.reasoning_config = snapshot["reasoning_config"]
|
|
primary = snapshot.get("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():
|
|
if "reasoning_config" in snapshot:
|
|
agent.reasoning_config = snapshot["reasoning_config"]
|
|
return
|
|
except Exception:
|
|
logger.debug("TUI one-turn model restore via primary runtime failed", exc_info=True)
|
|
if hasattr(agent, "switch_model"):
|
|
model, provider, api_key, base_url, api_mode = (snapshot.get(k, "") for k in _RUNTIME_KEYS)
|
|
agent.switch_model(
|
|
new_model=model, new_provider=provider, api_key=api_key, base_url=base_url,
|
|
api_mode=api_mode, capabilities=snapshot.get("capabilities"))
|
|
if "reasoning_config" in snapshot:
|
|
agent.reasoning_config = snapshot["reasoning_config"]
|
|
|
|
|
|
def _profile_runtime_scope_tokens(profile_home, *, hydrate_secrets: bool = True) -> _TurnScopes:
|
|
"""Bind HERMES_HOME + secret + terminal scope for ``profile_home`` (None = launch profile) and
|
|
return the reset tokens. The launch profile's SECRET scope is always bound — its ``.env`` over
|
|
the launch env (live while single-profile, frozen at activation afterwards; never live
|
|
``os.environ`` once a secondary context may have written to it, #107422) — so the credential
|
|
source is fixed at entry and an in-flight launch body survives a concurrent first-secondary
|
|
activation instead of hitting ``UnscopedSecretError`` mid-request. Its terminal policy is bound
|
|
only once multiplexing is active: single-profile terminal execution keeps the standalone
|
|
``os.environ`` bridge.
|
|
|
|
``hydrate_secrets=False`` skips resolving the profile's EXTERNAL secret sources (``op run``,
|
|
``bws``, ``sh -c`` — a subprocess with a 30s CLI budget holding a process-global lock). Pass it
|
|
from any body that persists state rather than calls a provider: the exit-flush worker has a 5s
|
|
TOTAL budget, and one slow source there costs the transcript the flush exists to save.
|
|
"""
|
|
from agent.secret_scope import is_multiplex_active
|
|
scopes = _TurnScopes()
|
|
try:
|
|
if profile_home:
|
|
home = Path(profile_home)
|
|
# External sources first: the requested profile may never have been served in this process.
|
|
if hydrate_secrets:
|
|
from hermes_cli.env_loader import hydrate_profile_secret_sources
|
|
hydrate_profile_secret_sources(home)
|
|
secrets = build_profile_secret_scope(home)
|
|
overlay = None
|
|
scopes.home = set_hermes_home_override(str(home))
|
|
else:
|
|
# The launch home IS get_hermes_home() (``_profile_home`` answers None for "already the
|
|
# launch profile"); single-profile, only its secrets need binding. Once multiplexing is
|
|
# active the override is bound too: an unset override is the "unbound context" signal
|
|
# plugin runtime bindings and per-home slots fail closed on (#118538).
|
|
# Resolve at call time like the launch state.db handle: a harness that
|
|
# re-homes the process after import must not read the old home's .env.
|
|
from tui_gateway.launch_profile_policy import launch_secret_scope, launch_terminal_env
|
|
home = _launch_home()
|
|
secrets = launch_secret_scope(home)
|
|
# No home stamp: this IS the process's own profile, and the stamp exists only to
|
|
# mark a FOREIGN home for serves_routed_profile().
|
|
scopes.secret = set_secret_scope(secrets)
|
|
if not is_multiplex_active():
|
|
return scopes
|
|
scopes.home = set_hermes_home_override(str(home))
|
|
overlay = launch_terminal_env()
|
|
if scopes.secret is None:
|
|
scopes.secret = set_secret_scope(secrets, profile_home=str(home) if profile_home else None)
|
|
# Same terminal policy the gateway binds per turn: a docker-configured profile
|
|
# must never resolve the launch process's pinned env. Failure → refusal scope.
|
|
from tools.terminal_scope import install_profile_terminal_scope
|
|
scopes.terminal = install_profile_terminal_scope(home, env_overlay=overlay)
|
|
return scopes
|
|
except Exception:
|
|
# A raise mid-bind leaves no return value for the caller to release —
|
|
# undo whatever was bound before propagating.
|
|
_release_profile_runtime_scope_tokens(scopes)
|
|
raise
|
|
|
|
|
|
def _release_profile_runtime_scope_tokens(scopes: _TurnScopes | None) -> None:
|
|
"""Release terminal → secret → home. Each reset is independent: a failing terminal reset must
|
|
not leave the previous profile's secrets / HERMES_HOME installed for the next body in this
|
|
context (a fail-open scope leak on the teardown path). The first failure is re-raised after
|
|
every scope has been released."""
|
|
if scopes is None:
|
|
return
|
|
from tools.terminal_scope import reset_terminal_scope
|
|
first_error: BaseException | None = None
|
|
for token, reset in ((scopes.terminal, reset_terminal_scope), (scopes.secret, reset_secret_scope),
|
|
(scopes.home, reset_hermes_home_override)):
|
|
if token is None:
|
|
continue
|
|
try:
|
|
reset(token)
|
|
except Exception as exc:
|
|
first_error = first_error or exc
|
|
if first_error is not None:
|
|
raise first_error
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _session_profile_runtime_scope(session: dict, *, hydrate_secrets: bool = True):
|
|
"""Bind model resolution to the session's profile config and secrets (launch profile included
|
|
once the process multiplexes; see ``_profile_runtime_scope_tokens``)."""
|
|
scopes = _profile_runtime_scope_tokens(session.get("profile_home"), hydrate_secrets=hydrate_secrets)
|
|
try:
|
|
yield
|
|
finally:
|
|
_release_profile_runtime_scope_tokens(scopes)
|
|
|
|
|
|
def _session_default_route(session: dict) -> tuple[str, str]:
|
|
"""``(model, provider)`` a not-yet-built session of its OWN profile will run on. Bare
|
|
``_resolve_startup_runtime()`` reads the LAUNCH profile's config, so a secondary session's reply or
|
|
first state.db row carried the launch profile's model id. On the Nous free tier the agent build pins
|
|
``nous/welcome`` (``pin_model_for_route``), so the configured default (often the silent default,
|
|
with no provider) is not what the session runs; report the pinned route instead."""
|
|
from hermes_cli.anon_auth import GUEST_MODEL, free_tier_route
|
|
with _session_profile_runtime_scope({"profile_home": session.get("profile_home") or None},
|
|
hydrate_secrets=False):
|
|
if not _resolve_startup_runtime()[1] and free_tier_route():
|
|
return GUEST_MODEL, "nous"
|
|
# Off the free tier, the model id alone: the provider is resolved when the agent is built.
|
|
return _resolve_model(), ""
|
|
|
|
|
|
def _lazy_info_route(session: dict, override: dict) -> dict:
|
|
"""``session.info``'s model and provider for a not-yet-built session: the client's sticky pick when it
|
|
sent one (so the client does not clobber it), else the profile's default route."""
|
|
model, provider = (override.get("model"), override.get("provider")) if override else _session_default_route(session)
|
|
return {"model": model, **({"provider": provider} if provider else {})}
|
|
|
|
|
|
def _restart_completed_failed_agent_build(sid: str, session: dict, failed_ready: threading.Event | None) -> bool:
|
|
"""Replace one completed failed build generation and start its retry."""
|
|
if failed_ready is None:
|
|
return False
|
|
with session.setdefault("agent_build_lock", threading.Lock()):
|
|
if (session.get("agent") is not None or session.get("agent_error") is None
|
|
or session.get("agent_ready") is not failed_ready or not failed_ready.is_set()):
|
|
return False
|
|
model_override = session.get("model_override")
|
|
resume_overrides = session.get("resume_runtime_overrides")
|
|
if isinstance(model_override, dict) and isinstance(resume_overrides, dict):
|
|
resume_overrides = {**resume_overrides, "model_override": model_override}
|
|
if provider := model_override.get("provider"):
|
|
resume_overrides["provider_override"] = provider
|
|
else:
|
|
resume_overrides.pop("provider_override", None)
|
|
session["resume_runtime_overrides"] = resume_overrides
|
|
session["agent_error"] = None
|
|
session["agent_ready"] = threading.Event()
|
|
session.pop("agent_build_started", None)
|
|
session.pop("_agent_build_thread", None)
|
|
_start_agent_build(sid, session)
|
|
return True
|
|
|
|
|
|
def _switch_request(raw_input: str, parsed_flags, persist_override) -> tuple[str, str, bool, bool, str]:
|
|
"""Normalize /model flags → (model_input, explicit_provider, one_turn, persist_global, reasoning_effort)."""
|
|
from hermes_cli.model_switch import (
|
|
MODEL_SWITCH_ERR_ONCE_WITH_GLOBAL, MODEL_SWITCH_ERROR_TEXT, parse_model_switch_args,
|
|
resolve_persist_behavior)
|
|
|
|
f = parse_model_switch_args(raw_input) if parsed_flags is None else parsed_flags
|
|
model_input, explicit_provider, is_global_flag, is_session, one_turn = (
|
|
f.model_input, f.explicit_provider, f.is_global, f.is_session, f.is_once)
|
|
# Conflict validation is the shared parser's; surface it with the canonical copy.
|
|
for code in getattr(f, "errors", ()):
|
|
raise ValueError(MODEL_SWITCH_ERROR_TEXT[code])
|
|
if is_global_flag and one_turn:
|
|
raise ValueError(MODEL_SWITCH_ERROR_TEXT[MODEL_SWITCH_ERR_ONCE_WITH_GLOBAL])
|
|
if persist_override is None:
|
|
persist_override = resolve_persist_behavior(
|
|
is_global_flag, is_session, is_once=one_turn, explicit_provider=explicit_provider)
|
|
if not model_input:
|
|
raise ValueError("model value required")
|
|
return model_input, explicit_provider, one_turn, persist_override, getattr(f, "reasoning_effort", "") or ""
|
|
|
|
|
|
def _current_model_runtime(agent, explicit_provider: str) -> tuple:
|
|
"""(provider, model, base_url, api_key) to switch from: live agent, else configured runtime."""
|
|
if agent:
|
|
return tuple(
|
|
getattr(agent, k, "") or "" for k in ("provider", "model", "base_url", "api_key"))
|
|
current_model = _resolve_model()
|
|
if explicit_provider:
|
|
return explicit_provider.strip(), current_model, "", ""
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider
|
|
runtime = resolve_runtime_provider(requested=None, target_model=current_model or None)
|
|
# Keep a callable api_key (Azure Entra bearer) unchanged: ``str()`` would
|
|
# yield "<function ...>" and poison switch_model validation.
|
|
key = runtime.get("api_key", "")
|
|
if not (callable(key) and not isinstance(key, str)):
|
|
key = str(key or "")
|
|
provider = str(runtime.get("provider", "") or "")
|
|
return provider, current_model, str(runtime.get("base_url", "") or ""), key
|
|
|
|
|
|
def _switch_away_provider(agent, explicit_provider: str, current_provider: str) -> str | None:
|
|
"""The provider of the model the user leaves. Agent-less with ``--provider``, ``current_provider``
|
|
is the TARGET (what switch_model wants), so the launch route names it instead; None when only a
|
|
credential resolve could tell, which the metric reads as ``unknown`` and so reports the model as
|
|
``custom``."""
|
|
if agent or not explicit_provider:
|
|
return current_provider
|
|
if env_provider := os.environ.get("HERMES_TUI_PROVIDER", "").strip():
|
|
return env_provider
|
|
if _env_model_seed():
|
|
return None
|
|
return _config_model_target()[1] or None
|
|
|
|
|
|
def _merge_preflight_warning(result, agent, session: dict, cfg, custom_provs) -> None:
|
|
"""Fold the context-compression preflight warning into ``result`` (best-effort)."""
|
|
try:
|
|
from hermes_cli.context_switch_guard import merge_preflight_compression_warning
|
|
cfg_ctx = None
|
|
mc = cfg.get("model", {}) if isinstance(cfg, dict) else None
|
|
if isinstance(mc, dict) and mc.get("context_length") is not None:
|
|
cfg_ctx = int(mc["context_length"])
|
|
merge_preflight_compression_warning(
|
|
result, agent=agent, messages=list(session.get("history", [])),
|
|
custom_providers=custom_provs, config_context_length=cfg_ctx)
|
|
except Exception as exc:
|
|
logger.debug("preflight-compression switch warning failed: %s", exc)
|
|
|
|
|
|
def _expensive_model_confirm(result, current_base_url: str, current_api_key, agent=None) -> dict | None:
|
|
"""Deferred-confirm response when the selection guards flag the target model (or, with a live
|
|
``agent``, the switch itself — large cached context), else None."""
|
|
try:
|
|
from hermes_cli.model_selection_guards import (
|
|
combined_selection_warning, selection_context_for_agent)
|
|
warning = combined_selection_warning(
|
|
result.new_model, provider=result.target_provider, base_url=result.base_url or current_base_url,
|
|
api_key=result.api_key or current_api_key, model_info=result.model_info,
|
|
selection_context=selection_context_for_agent(agent))
|
|
except Exception:
|
|
warning = None
|
|
if warning is None:
|
|
return None
|
|
msg = f"{warning.message}\n\n{result.warning_message}" if result.warning_message else warning.message
|
|
# Same contract as _set_model's deferred branch: confirm_message is canonical, warning legacy.
|
|
return {"value": result.new_model, "warning": msg, "confirm_required": True, "confirm_message": msg}
|
|
|
|
|
|
def _commit_agent_switch(sid: str, session: dict, agent, result, current_model: str, snapshot):
|
|
"""Swap the live agent in place, then restart/persist/mark/announce; a failed swap aborts."""
|
|
try:
|
|
agent.switch_model(
|
|
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))
|
|
except Exception as exc:
|
|
# The in-place swap rolled the agent back and re-raised. Abort the whole commit (worker
|
|
# restart, persist, marker, override, config write) or the session pins a broken model.
|
|
# Abort the commit: do NOT restart the slash worker, persist runtime, append the switch marker, set
|
|
# a session model_override, or persist to config — all of which would otherwise leave the session
|
|
# pinned to a broken model and kill the conversation on the next turn (#50163). A failed switch is a
|
|
# no-op; surface a clean error to the client.
|
|
logger.warning("In-place model switch failed for TUI agent: %s", exc)
|
|
raise ValueError(f"Model switch to {result.new_model} failed ({exc}); "
|
|
f"staying on {getattr(agent, 'model', current_model)}.") from exc
|
|
_restart_slash_worker(sid, session)
|
|
_persist_live_session_runtime(session)
|
|
_persist_live_session_system_prompt(session)
|
|
_append_model_switch_marker(session, model=result.new_model, provider=result.target_provider)
|
|
_emit_session_info(sid, session)
|
|
if snapshot is not None:
|
|
session["one_turn_model_restore"] = snapshot
|
|
else:
|
|
session.pop("one_turn_model_restore", None)
|
|
|
|
|
|
LARGE_MODEL_NOTICE_KEY = "model.large_notice"
|
|
|
|
|
|
def _apply_model_switch(
|
|
sid: str, session: dict, raw_input: str, *, confirm_expensive_model: bool = False,
|
|
pin_session_override: bool = True, parsed_flags: Any | None = None,
|
|
persist_override: bool | None = None, count_switch: bool = True) -> dict:
|
|
"""``count_switch=False``: an internal swap (config adoption, MoA one-shot and its restore), not a
|
|
user's /model pick, so it stays out of the shared-metrics switch count."""
|
|
from hermes_cli.model_switch import parse_model_switch_args, switch_model
|
|
if parsed_flags is None:
|
|
parsed_flags = parse_model_switch_args(raw_input)
|
|
if getattr(parsed_flags, "is_reset", False):
|
|
return _reset_session_model_override(sid, session)
|
|
model_input, explicit_provider, one_turn, persist_global, reasoning_effort = _switch_request(
|
|
raw_input, parsed_flags, persist_override)
|
|
if persist_global: # refused before switch_model() touches the live session
|
|
_refuse_managed_model_pick(reasoning_effort)
|
|
agent = session.get("agent")
|
|
if one_turn and not agent:
|
|
raise ValueError("/model --once requires a live session")
|
|
current_provider, current_model, current_base_url, current_api_key = _current_model_runtime(
|
|
agent, explicit_provider)
|
|
# User-defined providers let switch_model resolve named custom endpoints
|
|
# (e.g. "ollama-launch") and validate against saved model lists.
|
|
user_provs = custom_provs = cfg = None
|
|
with contextlib.suppress(Exception):
|
|
from hermes_cli.config import get_compatible_custom_providers, load_config
|
|
cfg = load_config()
|
|
user_provs = cfg.get("providers")
|
|
custom_provs = get_compatible_custom_providers(cfg)
|
|
result = switch_model(
|
|
raw_input=model_input, current_provider=current_provider, current_model=current_model,
|
|
current_base_url=current_base_url, current_api_key=current_api_key, is_global=persist_global,
|
|
explicit_provider=explicit_provider, user_providers=user_provs,
|
|
custom_providers=custom_provs)
|
|
if not result.success:
|
|
raise ValueError(result.error_message or "model switch failed")
|
|
if persist_global: # the exact plan, incl. conditional clears: still before any confirm or commit
|
|
_refuse_managed_model_pick(reasoning_effort, result)
|
|
restore_snapshot = _snapshot_agent_model_runtime(agent) if (one_turn and agent) else None
|
|
if agent:
|
|
_merge_preflight_warning(result, agent, session, cfg, custom_provs)
|
|
if not confirm_expensive_model:
|
|
confirm = _expensive_model_confirm(result, current_base_url, current_api_key, agent)
|
|
if confirm is not None:
|
|
return confirm
|
|
records_composer_override = (
|
|
pin_session_override and isinstance(session, dict) and not one_turn
|
|
and not persist_global and session.get("follow_profile_config"))
|
|
had_composer_profile = "composer_override_profile" in session
|
|
previous_composer_profile = session.get("composer_override_profile")
|
|
if records_composer_override:
|
|
profile_model, profile_provider = _config_model_target()
|
|
session["composer_override_profile"] = {
|
|
"model": profile_model, "provider": profile_provider}
|
|
try:
|
|
if agent:
|
|
# Provenance must exist before this transaction persists the switched runtime.
|
|
_commit_agent_switch(sid, session, agent, result, current_model, restore_snapshot)
|
|
except Exception:
|
|
if records_composer_override:
|
|
if had_composer_profile:
|
|
session["composer_override_profile"] = previous_composer_profile
|
|
else:
|
|
session.pop("composer_override_profile", None)
|
|
raise
|
|
# PER-SESSION override so a rebuild of THIS session (/new, resume) re-derives the model.
|
|
# Deliberately NOT written to process-global env (HERMES_MODEL & co.): the desktop hosts
|
|
# every same-profile session in one process, so os.environ would leak the switch to all.
|
|
if pin_session_override and isinstance(session, dict) and not one_turn:
|
|
session["model_override"] = {
|
|
"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:
|
|
from hermes_cli.model_switch import persist_model_selection
|
|
persist_model_selection(result)
|
|
if reasoning_effort:
|
|
_apply_switch_reasoning(sid, session, agent, reasoning_effort, persist_global=persist_global, one_turn=one_turn)
|
|
if count_switch:
|
|
from hermes_cli.observability.shared_metrics_events import record_model_switch
|
|
|
|
record_model_switch(
|
|
from_provider=_switch_away_provider(agent, explicit_provider, current_provider),
|
|
to_provider=result.target_provider, surface=_session_source(session), from_model=current_model,
|
|
session_id=getattr(agent, "session_id", None))
|
|
if (notice := getattr(result, "usage_notice", "")) and sid:
|
|
# Window plans, Large model: a warning, never a confirm. The existing notice event, so desktop
|
|
# (toast) and TUI (status line) show it for every switch path, live or applied at turn start.
|
|
_emit("notification.show", sid, {
|
|
"text": f"⚠ {notice}", "level": "warn", "kind": "ttl", "ttl_ms": 8000,
|
|
"key": LARGE_MODEL_NOTICE_KEY, "id": LARGE_MODEL_NOTICE_KEY})
|
|
return {
|
|
"value": result.new_model, "warning": result.warning_message or "",
|
|
"confirm_required": False,
|
|
"scope": "once" if one_turn else ("global" if persist_global else "session")}
|
|
|
|
|
|
def _reset_session_model_override(sid: str, session: dict) -> dict:
|
|
"""``/model reset`` (typed, ``config.set model``, Desktop's slash mirror, or a pick queued
|
|
mid-turn): drop this session's pin and any armed ``--once`` restore, adopt config.yaml's model
|
|
through the config-sync switch (no new pin, never written back), and persist the row so
|
|
``session.resume`` cannot restore the old pin. History is kept."""
|
|
from agent.i18n import t
|
|
had_override = bool(session.get("model_override"))
|
|
session.pop("model_override", None)
|
|
session.pop("one_turn_model_restore", None)
|
|
session.pop("config_model_seen", None) # the pinned sync recorded the target without adopting it
|
|
if "composer_override_profile" in session:
|
|
session["composer_override_profile"] = None # persisted as a removal
|
|
model, provider = _config_model_target()
|
|
agent = session.get("agent")
|
|
if model and agent is not None and (
|
|
model != getattr(agent, "model", "") or (provider and provider != getattr(agent, "provider", ""))):
|
|
_apply_model_switch(
|
|
sid, session, f"{model} --provider {provider}" if provider else model,
|
|
confirm_expensive_model=True, pin_session_override=False, persist_override=False,
|
|
count_switch=False)
|
|
_persist_live_session_runtime(session)
|
|
_emit_session_info(sid, session)
|
|
shown = model or getattr(agent, "model", "") or t("gateway.shared.unknown_value")
|
|
return {"value": shown, "warning": t("gateway.model.reset_done" if had_override else "gateway.model.reset_none",
|
|
model=shown),
|
|
"confirm_required": False, "scope": "session"}
|
|
|
|
|
|
def _apply_switch_reasoning(sid: str, session, agent, effort: str, *, persist_global: bool, one_turn: bool) -> None:
|
|
"""``/model X --reasoning <level>``: the effort rides with the pick and shares its scope. Runs
|
|
AFTER ``agent.switch_model`` (which re-resolves ``reasoning_config`` from config.yaml, so an
|
|
earlier write would be clobbered). ``--once`` restores through ``one_turn_model_restore`` —
|
|
the snapshot's ``primary_runtime`` carries the pre-switch ``reasoning_config``."""
|
|
from hermes_constants import parse_reasoning_effort
|
|
parsed = parse_reasoning_effort(effort)
|
|
if parsed is None:
|
|
return
|
|
if agent is not None:
|
|
agent.reasoning_config = parsed
|
|
if one_turn or not isinstance(session, dict):
|
|
return
|
|
if persist_global:
|
|
_write_config_key("agent.reasoning_effort", effort)
|
|
session.pop("create_reasoning_override", None) # global wins; see _set_reasoning
|
|
else:
|
|
session["create_reasoning_override"] = parsed
|
|
if agent is not None:
|
|
_persist_live_session_runtime(session)
|
|
_emit_session_info(sid, session) # the switch's own emit predates the effort change
|
|
|
|
|
|
def _sync_bot_capabilities(sid: str, session: dict) -> None:
|
|
"""Rebuild a Bot Chat session's agent when its capability surface changed. Bot Chats are
|
|
eternal sessions with toolsets/MCP baked in at construction, so a capability edit would
|
|
otherwise wait for /new: fingerprint at turn start and on change swap in a fresh agent for
|
|
the SAME session (history is DB-backed)."""
|
|
agent = session.get("agent")
|
|
if agent is None:
|
|
return
|
|
try:
|
|
title = str(getattr(agent, "_session_title_hint", "") or "").strip()
|
|
if not title:
|
|
db, key = getattr(agent, "_session_db", None), session.get("session_key") or ""
|
|
title = str((db.get_session_title(key) if (db and key) else None) or "").strip()
|
|
if title != "Bot Chat":
|
|
return
|
|
from tools.bot_mode_probe import capability_fingerprint
|
|
current = capability_fingerprint(session.get("profile_home") or None)
|
|
if current == "unavailable":
|
|
return
|
|
seen = session.get("bot_caps_seen")
|
|
session["bot_caps_seen"] = current
|
|
if seen is None or seen == current:
|
|
return
|
|
except Exception:
|
|
return
|
|
try:
|
|
tokens = _set_session_context(sid, cwd=_session_cwd(session))
|
|
try:
|
|
new_agent = _rebuild_session_agent(sid, session, session_id=session["session_key"],
|
|
platform_override=_session_source(session))
|
|
finally:
|
|
_clear_session_context(tokens)
|
|
new_agent._session_title_hint = "Bot Chat"
|
|
_emit("notice", sid, {"message": "Capabilities updated — this bot's tools and prompt were refreshed."})
|
|
except Exception as e:
|
|
logger.warning("Bot capability sync failed for %s: %s", sid, e)
|
|
|
|
|
|
def _sync_agent_model_with_config(sid: str, session: dict) -> None:
|
|
"""Adopt a config.yaml model change at turn start (like gateways do per message). Sessions
|
|
pinned with /model keep their choice; a failed switch keeps the current model."""
|
|
agent = session.get("agent")
|
|
if agent is None:
|
|
return
|
|
target = _config_model_target()
|
|
if not target[0]:
|
|
return
|
|
seen = session.get("config_model_seen")
|
|
if target == seen:
|
|
return
|
|
superseded_pin = None
|
|
if session.get("model_override"):
|
|
composer_profile = session.get("composer_override_profile")
|
|
pinned_profile = (
|
|
str(composer_profile.get("model") or "").strip(),
|
|
str(composer_profile.get("provider") or "").strip(),
|
|
) if isinstance(composer_profile, dict) else None
|
|
if pinned_profile is None or pinned_profile == target:
|
|
return
|
|
# A later profile edit supersedes the canonical chat's explicit pick. Clearing both fields lets
|
|
# the normal config-sync path switch now and prevents the old composer pick resurfacing on rebuild.
|
|
superseded_pin = session.pop("model_override"), composer_profile
|
|
session["composer_override_profile"] = None
|
|
# Record first so a broken config gets one attempt per edit, not per turn.
|
|
session["config_model_seen"] = target
|
|
model, provider = target
|
|
# Already on the configured model (resumed before first sync, or a config revert after
|
|
# a failed switch): adopt without switching.
|
|
if model == getattr(agent, "model", "") and (not provider or provider == getattr(agent, "provider", "")):
|
|
if superseded_pin is not None:
|
|
_persist_live_session_runtime(session)
|
|
return
|
|
raw = f"{model} --provider {provider}" if provider else model
|
|
try:
|
|
# This sync ADOPTS a config.yaml change; it must never write config back (that is
|
|
# how `hermes --tui -m` once leaked into config.yaml).
|
|
_apply_model_switch(
|
|
sid, session, raw, confirm_expensive_model=True, pin_session_override=False,
|
|
persist_override=False, count_switch=False)
|
|
except Exception as e:
|
|
logger.warning("Configured model %s could not be adopted for session %s: %s", model, sid, e)
|
|
from gateway.warning_notifications import render_notification
|
|
render_notification(
|
|
lambda: _emit("error", sid, {"message": f"Could not switch to configured model {model}: {e}"}),
|
|
platform="tui", user_config=getattr(session.get("agent"), "_notification_config", None))
|
|
|
|
|
|
def _pending_switch_selection_warning(model: str, provider: str, agent: Any = None) -> str | None:
|
|
"""Selection-guard message for a model queued mid-turn, or ``None``. Runs BEFORE the pick is
|
|
stashed (the client can still turn the response into a confirm prompt); only pre-resolution
|
|
inputs exist for the model guards, so they can only under-fire — ``_apply_model_switch`` is the
|
|
backstop. The live ``agent`` supplies the context-cache guard's session size here, since at turn
|
|
start that guard can no longer ask and would drop the pick."""
|
|
if not model:
|
|
return None
|
|
try:
|
|
from hermes_cli.model_selection_guards import (
|
|
combined_selection_warning, selection_context_for_agent)
|
|
warning = combined_selection_warning(
|
|
model, provider=provider or None, selection_context=selection_context_for_agent(agent))
|
|
except Exception:
|
|
return None
|
|
return warning.message if warning is not None else None
|
|
|
|
|
|
def register(server) -> None:
|
|
"""Publish this module's helpers + handlers onto ``server``, rebound to its globals."""
|
|
bind_module(globals(), server, skip=("_",))
|