ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

refactor(tui_gateway): W4 wave2 pass 4 — profile/config_set docstring+comment compaction, mirror helpers

Browse Source
This commit is contained in:
Teknium
2026-09-03 02:36:23 -07:00
parent b7747dce42
commit 0548956da0
4 changed files with 58 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/methods_config_set.py
+19 -28
View File
@@ -1,5 +1,5 @@
"""``config.set`` — one JSON-RPC method, dispatched on ``key`` through a table. Bodies are
rebound onto server.py's globals (method_ctx.bind_module) and reference them bare. Each
"""``config.set`` — one JSON-RPC method, dispatched on ``key`` through ``_CONFIG_SETTERS``. Bodies
are rebound onto server.py's globals (method_ctx.bind_module) and reference them bare. Each
``_set_*`` takes ``(rid, params, key, value, session)`` and returns the JSON-RPC envelope.
Keys match exactly except ``details_mode.<section>`` (prefix) and ``_DISPLAY_TOGGLE_KEYS``.
"""
@@ -98,19 +98,17 @@ def _set_model(rid, params, key, value, session):
except Exception:
pending_model = str(value)
pending_provider = (getattr(parsed, "explicit_provider", "") or "").strip()
# Selection guards run HERE (the only moment a confirm round-trip is possible);
# otherwise an unconfirmed stashed pick is dropped at turn start. On a warning
# nothing is stashed; the client re-sends with confirm_expensive_model.
# Selection guards run HERE (the only moment a confirm round-trip is possible; an
# unconfirmed stashed pick is dropped at turn start). On a warning nothing is stashed.
# `confirm_message` is canonical, `warning` its legacy alias.
if not confirmed:
pending_warning = _pending_switch_selection_warning(pending_model, pending_provider)
if pending_warning is not None:
return _cfgset_model_ok(rid, key, pending_model, pending_warning, True,
pending_warning, "session", deferred=False)
# display_*: _session_info shows the user's pick while pending, not the live old model.
session["pending_model_switch"] = {
"raw": value, "confirm_expensive_model": confirmed,
# _session_info reports these while pending so the end-of-turn settle keeps
# showing the user's pick, not the still-live old model.
"display_model": pending_model, "display_provider": pending_provider}
return _cfgset_model_ok(rid, key, pending_model, "", False, "", "session", deferred=True)
parsed_flags = parse_model_switch_args(value)
@@ -179,9 +177,9 @@ def _set_fast(rid, params, key, value, session):
if overrides is None:
return _err(rid, 4002, "fast mode is not available for this model")
if session is not None:
# Session-scoped like `reasoning` (global persistence is `--global` / Settings → Model):
# writing config.yaml here flipped fast mode for every other surface. The create
# override keeps the choice across lazy builds and rebuilds; "" pins normal.
# Session-scoped like `reasoning` (global = `--global` / Settings → Model): writing
# config.yaml here flipped fast mode for every surface. The create override survives lazy
# builds and rebuilds; "" pins normal.
session["create_service_tier_override"] = {"fast": "priority", "normal": ""}.get(nv, nv)
else:
_write_config_key("agent.service_tier", nv)
@@ -223,8 +221,7 @@ def _set_verbose(rid, params, key, value, session):
def _set_focus(rid, params, key, value, session):
# Focus view (/focus): enabling stashes the configured tool_progress mode and pins it
# "off"; disabling restores the stash.
# /focus: enabling stashes the configured tool_progress mode and pins it "off"; disabling restores.
from hermes_cli.focus_view import FOCUS_TOOL_PROGRESS_MODE, normalize_tool_progress_mode, resolve_focus_arg
d_f = _display_cfg()
cur_focus = bool(d_f.get("focus_view", False))
@@ -261,7 +258,7 @@ def _set_approval_mode(rid, params, key, value, session):
@_cfgset_guarded
def _set_yolo(rid, params, key, value, session):
# scope="session" (default; Shift+Tab) toggles ONLY this session's flag. scope="global"
# scope="session" (default; Shift+Tab) toggles ONLY this session's flag; scope="global"
# (Shift+click the zap) flips persistent approvals.mode between "off" and "manual".
scope = _word(params.get("scope") or "session")
from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled
@@ -274,8 +271,7 @@ def _set_yolo(rid, params, key, value, session):
appr = _load_cfg().get("approvals")
appr = appr if isinstance(appr, dict) else {}
enable = _resolve_toggle(_normalize_approval_mode(appr.get("mode", "manual")) == "off")
# Binary affordance: no restore of a prior "smart"/custom mode (those live in config.yaml).
_write_config_key("approvals.mode", "off" if enable else "manual")
_write_config_key("approvals.mode", "off" if enable else "manual") # binary: no "smart" restore
_emit_all_session_info() # reflect the flip in every live indicator
elif session:
skey = session["session_key"]
@@ -318,9 +314,7 @@ def _set_reasoning(rid, params, key, value, session):
_write_config_key("agent.reasoning_effort", arg)
if session is not None:
session.pop("create_reasoning_override", None)
else:
# Session-scoped like the gateway's `/reasoning <level>`; otherwise every desktop
# model-menu pick rewrote the global default.
else: # session-scoped like the gateway's `/reasoning <level>`; a menu pick must not rewrite the global
session["create_reasoning_override"] = parsed
if session and session.get("agent") is not None:
session["agent"].reasoning_config = parsed
@@ -338,8 +332,8 @@ def _set_details_mode(rid, params, key, value, session):
def _set_details_section(rid, params, key, value, session):
# `details_mode.<section>` -> `display.sections.<section>`; empty clears the override so the
# frontend applies built-in section defaults before the global details_mode.
# `details_mode.<section>` -> `display.sections.<section>`; empty clears the override (frontend
# then applies built-in section defaults before the global details_mode).
section = key.split(".", 1)[1]
if section not in _DETAIL_SECTION_NAMES:
return _err(rid, 4002, f"unknown section: {section}")
@@ -355,8 +349,7 @@ def _set_thinking_mode(rid, params, key, value, session):
if nv not in {"collapsed", "truncated", "full"}:
return _err(rid, 4002, f"unknown thinking_mode: {value}")
_write_config_key("display.thinking_mode", nv)
# Backward compatibility bridge: keep details_mode aligned.
_write_config_key("display.details_mode", "expanded" if nv == "full" else "collapsed")
_write_config_key("display.details_mode", "expanded" if nv == "full" else "collapsed") # compat bridge
return _kv(rid, key, nv)
@@ -372,7 +365,8 @@ def _toggle_setter(rid, key, value, raw, aliases: dict, flipped, cfg_key: str, r
# on/off/toggle display booleans: key -> (display field, accepted word -> bool).
_DISPLAY_BOOLS = {
"density": ("tui_compact", {"on": True, "off": False}),
"battery": ("battery", {"on": True, "true": True, "yes": True, "off": False, "false": False, "no": False})}
"battery": ("battery", {"on": True, "true": True, "yes": True, "off": False, "false": False,
"no": False})}
def _set_display_bool(rid, params, key, value, session):
@@ -437,8 +431,7 @@ def _set_prompt_like(rid, params, key, value, session):
_save_cfg(cfg)
elif key == "personality":
pname, new_prompt = _validate_personality(str(value or ""), cfg)
# Personality persists through hermes_cli.personality (single owner), never the
# user-owned global system prompt.
# Persists via hermes_cli.personality (single owner), never the user-owned system prompt.
from hermes_cli.personality import persist_personality
persist_personality(pname)
resp["value"] = str(value or "none")
@@ -448,9 +441,7 @@ def _set_prompt_like(rid, params, key, value, session):
resp["info"] = info
else:
_write_config_key(f"display.{key}", value)
if key == "skin":
# Every surface repaints; sync the watcher baseline so the poll loop doesn't
# re-broadcast the skin this RPC just applied.
if key == "skin": # every surface repaints; sync the watcher baseline (no re-broadcast)
_broadcast_global_event("skin.changed", resolve_skin())
_note_skin_broadcast()
return _ok(rid, resp)
tui_gateway/methods_images.py
+3 -2
View File
@@ -55,8 +55,9 @@ def _(rid, params: dict) -> dict:
if is_truthy_value(params.get("probe", False)):
return _ok(rid, {"available": available})
if not available:
return _ok(rid, {"available": False, "success": False,
"error": "No image generation backend configured (run `hermes tools` to enable one)."})
return _ok(rid, {
"available": False, "success": False,
"error": "No image generation backend configured (run `hermes tools` to enable one)."})
prompt = str(params.get("prompt") or "").strip()
if not prompt:
return _err(rid, 4071, "prompt required")
tui_gateway/methods_profiles.py
+27 -36
View File
@@ -1,8 +1,7 @@
"""Profile JSON-RPC handlers — the ws twin of the dashboard's /api/profiles (desktop plugins
only have the ws door), on the same `hermes_cli.profiles` primitives.
Bodies are rebound onto server.py's globals (method_ctx.bind_module) and use them bare;
module-level names are published onto server.py, so they must not collide with its globals.
only have the ws door), on the same `hermes_cli.profiles` primitives. Bodies are rebound onto
server.py's globals (method_ctx.bind_module) and use them bare; module-level names are published
onto server.py, so they must not collide with its globals.
"""
import contextlib
@@ -21,7 +20,6 @@ _ASSET_MAGIC = {"png": [(0, 8, b"\x89PNG\r\n\x1a\n")], "jpg": [(0, 3, b"\xff\xd8
def _profile_handler(name: str, code: int):
"""``@method(name)`` whose body's uncaught exception becomes ``_err(rid, code, str(e))``."""
def deco(fn):
def handler(rid, params: dict) -> dict:
try:
@@ -47,7 +45,7 @@ def _model_provider_params(params) -> tuple:
def _try(fn, default):
"""``fn()`` or ``default`` on any exception — best-effort sections must never fail each other."""
"""``fn()`` or ``default`` on any exception (best-effort sections must never fail each other)."""
try:
return fn()
except Exception:
@@ -55,7 +53,6 @@ def _try(fn, default):
def _best_effort(fn) -> bool:
"""Run ``fn``; True on success, False on any exception."""
return _try(lambda: (fn(), True)[1], False)
@@ -70,7 +67,7 @@ def _hermes_home_scope(path):
def _resolve_profile(rid, params):
"""``(name, profile_dir, err)`` — err is the 4063 (name required) / 4064 (not found) response."""
"""``(name, profile_dir, err)``; err = 4063 (name required) / 4064 (not found) response."""
name = str(params.get("name") or "").strip()
if not name:
return name, None, _err(rid, 4063, "name required")
@@ -98,7 +95,7 @@ def _clean_revisions(raw: dict) -> dict:
def _latest_message_preview(db, session_id):
"""≤80-char excerpt of the NEWEST active user/assistant message, or "" (roster semantics).
Same query shape as ``SessionDB.latest_message_row_id`` — keep them in step."""
Same query shape as ``SessionDB.latest_message_row_id``; keep them in step."""
try:
with db._lock:
row = db._conn.execute(
@@ -117,7 +114,7 @@ def _latest_message_preview(db, session_id):
def _resurrect_recoverable_canonical(db, profile_path, session_id):
"""Un-archive an accidentally archived canonical row (judged read-only, written via a
short-lived writable handle), or False."""
short-lived writable handle); False otherwise."""
try:
row = db.get_session(session_id)
if not row or not row.get("archived"):
@@ -162,8 +159,8 @@ def _canonical_session_row(db, profile_path):
def _latest_profile_session_rows(db):
"""(newest human-facing session, newest worker session). The worker row lets rosters show
a profile as working (workers heartbeat ``last_activity_at`` every ≤60s)."""
"""(newest human-facing session, newest worker session); the worker row lets rosters show a
profile as working (workers heartbeat ``last_activity_at`` every ≤60s)."""
try:
human = worker = None
for s in db.list_sessions_rich(source=None, limit=20, order_by_last_active=True, compact_rows=True):
@@ -206,8 +203,7 @@ def _profile_session_fields(row, profile_path):
def _profile_ui_meta_fields(row: dict, profile_dir) -> None:
"""Attach ``ui_meta`` / ``ui_meta_revisions`` / ``has_avatar`` from profile.yaml + assets.
``ui_meta_revisions`` is always present: it feature-detects gateway-owned CAS even for a
brand-new profile."""
``ui_meta_revisions`` is always present: it feature-detects gateway-owned CAS for a new profile."""
row["ui_meta_revisions"] = {}
raw_meta = _read_profile_yaml(profile_dir)
ui_meta = raw_meta.get("ui_meta")
@@ -228,17 +224,15 @@ def _(rid, params: dict) -> dict:
include_sessions = is_truthy_value(params.get("include_sessions", True))
out = []
for p in list_profiles():
row = {
"name": p.name, "path": str(p.path), "is_default": bool(p.is_default),
"model": p.model, "provider": p.provider,
"description": p.description or "", "display_name": p.display_name or "",
"skill_count": p.skill_count or 0}
row = {"name": p.name, "path": str(p.path), "is_default": bool(p.is_default), "model": p.model,
"provider": p.provider, "description": p.description or "",
"display_name": p.display_name or "", "skill_count": p.skill_count or 0}
if include_sessions:
_profile_session_fields(row, p.path)
_profile_ui_meta_fields(row, Path(str(p.path)))
out.append(row)
# Capability flag: this backend injects the Bot Mode teammate-messaging
# protocol into every session, so clients must not append it to SOUL.md.
# bot_mode_protocol: this backend injects the Bot Mode teammate-messaging protocol into every
# session, so clients must not append it to SOUL.md.
return _ok(rid, {"profiles": out, "bot_mode_protocol": True})
@@ -276,7 +270,7 @@ def _mirror_auth(path, launch_home) -> bool:
def _mirror_voice_sections(path) -> bool:
"""Copy stt/tts/voice sections from the launch profile (a fresh profile has only ``model``,
so voice fell back to defaults); True if written. Canonical loaders under the home override."""
so voice fell back to defaults); True if written."""
try:
from hermes_cli.config import load_config_readonly, read_user_config_raw, save_config
src_cfg = load_config_readonly() or {}
@@ -288,8 +282,7 @@ def _mirror_voice_sections(path) -> bool:
dst_cfg = read_user_config_raw() or {}
missing = {k: v for k, v in sections.items() if k not in dst_cfg}
if missing:
dst_cfg.update(missing)
save_config(dst_cfg)
save_config({**dst_cfg, **missing})
return bool(missing)
except Exception:
return False
@@ -314,10 +307,9 @@ def _mirror_launch_credentials(path, params: dict) -> dict:
"""Copy launch .env / auth.json / voice sections into a new profile (best-effort per item).
``share_auth`` reports ``auth: "shared"`` and skips the auth copy; ``mirror_credentials``
false skips everything. ``model_inherited`` is filled in by the caller."""
mirrored = {"env": False, "auth": False, "model_inherited": False, "voice": False}
share_auth = is_truthy_value(params.get("share_auth", False))
if share_auth:
mirrored["auth"] = "shared"
mirrored = {"env": False, "auth": "shared" if share_auth else False, "model_inherited": False,
"voice": False}
if not is_truthy_value(params.get("mirror_credentials", True)):
return mirrored
launch_home = get_hermes_home()
@@ -332,9 +324,8 @@ def _mirror_launch_credentials(path, params: dict) -> dict:
def _(rid, params: dict) -> dict:
"""Create a profile (ws twin of POST /api/profiles). Params: ``name``, ``description``,
``clone_from`` (omitted = fresh + bundled skills), ``clone_all``, ``no_skills``, ``soul``,
``model`` + ``provider``, ``share_auth``, ``mirror_credentials`` (default true — a
``create_profile()`` seeds a comment-only .env and no auth.json, so a headless profile had
NO provider)."""
``model`` + ``provider``, ``share_auth``, ``mirror_credentials`` (default true: a bare
``create_profile()`` seeds a comment-only .env and no auth.json = NO provider headless)."""
name = str(params.get("name") or "").strip()
if not name:
return _err(rid, 4061, "name required")
@@ -371,7 +362,7 @@ def _(rid, params: dict) -> dict:
def _describe_toolsets(cfg):
"""``(toolsets, pinned_set)`` as the `hermes tools` checklist presents them (the raw registry
leaks platform composites and reports everything "enabled" without a pin)."""
leaks platform composites and reports everything enabled without a pin)."""
from hermes_cli.tools_config import (
_get_effective_configurable_toolsets, _get_platform_tools, _toolset_allowed_for_platform)
from toolsets import resolve_toolset
@@ -478,7 +469,7 @@ def _configure_ui_meta(profile_dir, params, applied) -> None:
def _configure_model(profile_dir, params, applied):
"""Apply a ``model`` + ``provider`` pin, or return a confirm message and write NOTHING (client
resends with ``confirm_expensive_model``). A failing guard = "no warning" (as _apply_model_switch)."""
resends with ``confirm_expensive_model``). A failing guard = no warning (as _apply_model_switch)."""
model, provider = _model_provider_params(params)
confirm_message = None
if not (model and provider):
@@ -531,7 +522,8 @@ def _configure_cfg_sections(profile_dir, params, applied) -> None:
# Launch catalog read BEFORE the home override flips config resolution.
launch_mcp = {}
if want_mcp:
launch_mcp = _try(lambda: (_lazy("hermes_cli.config", "load_config_readonly")() or {}).get("mcp_servers"), {})
load_launch = _lazy("hermes_cli.config", "load_config_readonly")
launch_mcp = _try(lambda: (load_launch() or {}).get("mcp_servers"), {})
launch_mcp = launch_mcp if isinstance(launch_mcp, dict) else {}
with _hermes_home_scope(profile_dir):
from hermes_cli.config import load_config, save_config
@@ -555,7 +547,7 @@ def _configure_cfg_sections(profile_dir, params, applied) -> None:
def _(rid, params: dict) -> dict:
"""Editor Save: ``name`` plus any of ``ui_meta`` (+ ``ui_meta_expected_revisions``), ``soul``,
``description``, ``model`` + ``provider`` (+ ``confirm_expensive_model``), ``disabled_skills``,
``enabled_toolsets``, ``enabled_mcp_servers``. Sections are independent; ``applied`` reports each."""
``enabled_toolsets``, ``enabled_mcp_servers``; sections are independent, ``applied`` reports each."""
_name, profile_dir, err = _resolve_profile(rid, params)
if err is not None:
return err
@@ -590,8 +582,7 @@ def _unlink_asset_files(assets_dir, asset) -> int:
@_profile_handler("profiles.set_asset", 5065)
def _(rid, params: dict) -> dict:
"""Store ``assets/<asset>.<ext>`` atomically. Params: ``name``, ``asset`` (``"avatar"`` only),
``data`` (data URL or base64; PNG/JPEG/WebP ≤2MB, sniffed by magic bytes — never the declared
mime) or ``clear: true``."""
``data`` (data URL or base64; PNG/JPEG/WebP ≤2MB, format sniffed) or ``clear: true``."""
asset = str(params.get("asset") or "avatar").strip().lower()
if not str(params.get("name") or "").strip():
return _err(rid, 4063, "name required")
tui_gateway/methods_voice.py
+9 -5
View File
@@ -478,7 +478,8 @@ def _(rid, params: dict) -> dict:
if not reqs["available"]:
logger.warning("wake.start(%s): not available — %s", surface, reqs.get("hint"))
return refused("unavailable", hint=reqs.get("hint") or "", capture=capture_mode)
enabled_persisted = bool(params.get("persist") and not cfg.get("enabled") and _persist_wake_enabled(True))
persist = bool(params.get("persist"))
enabled_persisted = bool(persist and not cfg.get("enabled") and _persist_wake_enabled(True))
if enabled_persisted:
cfg = {**cfg, "enabled": True}
if not wake_surface_enabled(surface, cfg):
@@ -489,15 +490,18 @@ def _(rid, params: dict) -> dict:
surface, reason, cfg.get("enabled"), cfg.get("surface"))
return refused(reason)
existing_owner, existing_surface = _wake_owner_snapshot()
if existing_owner is not None and (_transport_is_dead(existing_owner) or not owns_listener(existing_owner)):
if existing_owner is not None and (
_transport_is_dead(existing_owner) or not owns_listener(existing_owner)
):
_release_wake_for_transport(existing_owner)
existing_owner, existing_surface = None, ""
if existing_owner is not None and existing_owner is not transport:
return refused("owned", owner_surface=existing_surface)
try:
on_detect = _wake_detect_handler(transport, str(params.get("session_id") or ""), wake_phrase(cfg),
bool(cfg.get("start_new_session", True)))
start_listening(on_detect, owner=transport, config=cfg, external_audio=capture_mode == "client")
on_detect = _wake_detect_handler(transport, str(params.get("session_id") or ""),
wake_phrase(cfg), bool(cfg.get("start_new_session", True)))
start_listening(on_detect, owner=transport, config=cfg,
external_audio=capture_mode == "client")
except WakeWordInUse:
return refused("owned", owner_surface=existing_surface or None)
except Exception as e: