diff --git a/tui_gateway/methods_config.py b/tui_gateway/methods_config.py
index 140c7ab837..54fd12f8fa 100644
--- a/tui_gateway/methods_config.py
+++ b/tui_gateway/methods_config.py
@@ -53,12 +53,12 @@ def _(rid, params: dict) -> dict:
     policy = _repo_discovery_policy()
     policy_key = _repo_discovery_policy_key(policy)
     incoming_raw = params.get("discovery_policy")
-    incoming_policy = (
-        _repo_discovery_policy(incoming_raw) if isinstance(incoming_raw, dict) else None)
-    incoming_matches = (incoming_policy is not None
-                        and _repo_discovery_policy_key(incoming_policy) == policy_key)
-    accept_legacy_default = (incoming_policy is None
-                             and _repo_discovery_policy_is_default(policy))
+    incoming_policy = _repo_discovery_policy(incoming_raw) if isinstance(incoming_raw, dict) else None
+    if incoming_policy is not None:
+        accepted = _repo_discovery_policy_key(incoming_policy) == policy_key
+    else:
+        accepted = _repo_discovery_policy_is_default(policy)  # legacy client without a policy
+    accepted = bool(policy["enabled"] and accepted)
     pairs: list[tuple[str, str | None]] = []
     for item in params.get("repos") or []:
         if isinstance(item, str):
@@ -67,14 +67,12 @@ def _(rid, params: dict) -> dict:
             pairs.append((str(item["root"]), item.get("label")))
     with pdb.connect_closing() as conn:
         _reconcile_repo_discovery(pdb, conn, policy, policy_key)
-        accepted = bool(policy["enabled"] and (incoming_matches or accept_legacy_default))
         if accepted:
             pdb.record_discovered_repos(conn, pairs, replace=True, policy_key=policy_key)
         elif not policy["enabled"]:
             pdb.clear_discovered_repos(conn, policy_key=policy_key)
     with _profile_db(params) as db:
-        repos = ([] if db is None
-                 else _discover_repos_payload(db, include_cached=policy["enabled"]))
+        repos = [] if db is None else _discover_repos_payload(db, include_cached=policy["enabled"])
         return _ok(rid, {"repos": repos, "accepted": accepted, "discovery_policy": policy})
 
 
@@ -97,9 +95,8 @@ def _(rid, params: dict) -> dict:
         tree, active_id = _stamped_project_tree(
             db, params, preview_limit=int(params.get("preview_limit") or 3), hydrate=False,
             session_limit=int(params.get("session_limit") or 2000), include_discovered=True)
-        return _ok(rid, {
-            "projects": tree["projects"], "active_id": active_id,
-            "scoped_session_ids": tree["scoped_session_ids"]})
+        return _ok(rid, {"projects": tree["projects"], "active_id": active_id,
+                         "scoped_session_ids": tree["scoped_session_ids"]})
 
 
 @_projects_handler("projects.project_sessions")
@@ -115,61 +112,51 @@ def _(rid, params: dict) -> dict:
         tree, _active = _stamped_project_tree(
             db, params, preview_limit=0, hydrate=True,
             session_limit=int(params.get("session_limit") or 5000), include_discovered=False)
-        proj = next((p for p in tree["projects"] if p["id"] == project_id), None)
-        return _ok(rid, {"project": proj})
+        return _ok(rid, {"project": next((p for p in tree["projects"] if p["id"] == project_id), None)})
 
 
-# ── config.get — one getter per key; returns the result payload or a full ``_err`` response
-# (dicts containing "error" pass through untouched).
+# ── config.get — one getter per key returning the result payload.
+
+def _display_raw() -> dict:
+    return _load_cfg().get("display") or {}
 
 
-def _display_mode(cfg: dict, key: str, allowed: frozenset, default: str) -> str:
-    raw = str((cfg.get("display") or {}).get(key, default) or default).strip().lower()
+def _display_word(key: str, default: str, allowed) -> str:
+    """Normalised ``display.<key>``; unknown/garbage values read back as ``default``."""
+    raw = str(_display_raw().get(key, default) or "").strip().lower()
     return raw if raw in allowed else default
 
 
 _THINKING_MODES = frozenset({"collapsed", "truncated", "full"})
 
 
-def _cfg_get_provider(rid, params):
-    try:
-        from hermes_cli.models import list_available_providers, normalize_provider
-        model = _resolve_model()
-        parts = model.split("/", 1)
-        return {
-            "model": model,
-            "provider": normalize_provider(parts[0]) if len(parts) > 1 else "unknown",
+def _cfg_get_provider(params):
+    from hermes_cli.models import list_available_providers, normalize_provider
+    model = _resolve_model()
+    parts = model.split("/", 1)
+    return {"model": model, "provider": normalize_provider(parts[0]) if len(parts) > 1 else "unknown",
             "providers": list_available_providers()}
-    except Exception as e:
-        return _err(rid, 5013, str(e))
 
 
-def _cfg_get_profile(rid, params):
+def _cfg_get_profile(params):
     from hermes_constants import display_hermes_home
     return {"home": str(_hermes_home), "display": display_hermes_home()}
 
 
-def _cfg_get_project(rid, params):
+def _cfg_get_project(params):
     cfg_terminal = _load_cfg().get("terminal") or {}
     raw = str(params.get("cwd", "") or cfg_terminal.get("cwd", "") or "").strip()
     cwd = _completion_cwd({"cwd": raw} if raw else {})
     return {"cwd": cwd, "branch": _git_branch_for_cwd(cwd)}
 
 
-def _cfg_get_indicator(rid, params):
-    # Normalize so a hand-edited config.yaml (stray casing / unknown value) reads back the SAME
-    # value the TUI rendered (frontend falls back to DEFAULT_INDICATOR_STYLE for the same inputs).
-    norm = str((_load_cfg().get("display") or {}).get("tui_status_indicator", "")).strip().lower()
-    return {"value": norm if norm in INDICATOR_STYLES else DEFAULT_INDICATOR_STYLE}
-
-
-def _cfg_get_personality(rid, params):
+def _cfg_get_personality(params):
     # EFFECTIVE personality via the single owner — a stale/unknown name must not show as active.
     from hermes_cli.personality import active_personality_name
     return {"value": active_personality_name(_load_cfg()) or "none"}
 
 
-def _cfg_get_reasoning(rid, params):
+def _cfg_get_reasoning(params):
     cfg = _load_cfg()
     session = _sessions.get(params.get("session_id", ""))
     reasoning_config = None
@@ -188,7 +175,7 @@ def _cfg_get_reasoning(rid, params):
     return {"value": effort, "display": display}
 
 
-def _cfg_get_fast(rid, params):
+def _cfg_get_fast(params):
     # `config.set fast` is session-scoped: prefer the session's live/pinned value over the
     # global key (a pre-build session keeps its pin in create_service_tier_override).
     session = _sessions.get(params.get("session_id", ""))
@@ -204,28 +191,15 @@ def _cfg_get_fast(rid, params):
     return {"value": "fast" if tier == "priority" else "normal"}
 
 
-def _cfg_get_approval_mode(rid, params):
-    try:
-        return {"value": _load_approval_mode()}
-    except Exception as e:
-        return _err(rid, 5001, str(e))
-
-
-def _cfg_get_thinking_mode(rid, params):
-    cfg = _load_cfg()
-    raw = str((cfg.get("display") or {}).get("thinking_mode", "") or "").strip().lower()
-    if raw in _THINKING_MODES:
+def _cfg_get_thinking_mode(params):
+    raw = _display_word("thinking_mode", "", _THINKING_MODES)
+    if raw:
         return {"value": raw}
-    dm = _display_mode(cfg, "details_mode", _DETAIL_MODES, "collapsed")
+    dm = _display_word("details_mode", "collapsed", _DETAIL_MODES)
     return {"value": "full" if dm == "expanded" else "collapsed"}
 
 
-def _cfg_get_theme(rid, params):
-    raw = str(_display_cfg().get("tui_theme", "auto")).strip().lower()
-    return {"value": raw if raw in {"auto", "light", "dark"} else "auto"}
-
-
-def _cfg_get_mtime(rid, params):
+def _cfg_get_mtime(params):
     cfg_path = _hermes_home / "config.yaml"
     try:
         mtime = cfg_path.stat().st_mtime if cfg_path.exists() else 0
@@ -236,55 +210,53 @@ def _cfg_get_mtime(rid, params):
     return {"mtime": mtime, "mcp_rev": _compute_mcp_rev()}
 
 
-def _config_getters() -> dict:
-    """key -> getter(rid, params). Built per call so, once rebound onto server.py, every entry
-    resolves to the rebound helper copies rather than this module's originals."""
-    return {
-        "provider": _cfg_get_provider,
-        "profile": _cfg_get_profile,
-        "project": _cfg_get_project,
-        "full": lambda rid, params: {"config": _load_cfg()},
-        "prompt": lambda rid, params: {"prompt": _load_cfg().get("custom_prompt", "")},
-        "skin": lambda rid, params: {"value": (_load_cfg().get("display") or {}).get("skin", "default")},
-        "indicator": _cfg_get_indicator,
-        "personality": _cfg_get_personality,
-        "reasoning": _cfg_get_reasoning,
-        "fast": _cfg_get_fast,
-        "busy": lambda rid, params: {"value": _load_busy_input_mode()},
-        "approval_mode": _cfg_get_approval_mode,
-        "approvals.mode": _cfg_get_approval_mode,
-        "details_mode": lambda rid, params: {
-            "value": _display_mode(_load_cfg(), "details_mode", _DETAIL_MODES, "collapsed")},
-        "thinking_mode": _cfg_get_thinking_mode,
-        "density": lambda rid, params: {
-            "value": "on" if bool((_load_cfg().get("display") or {}).get("tui_compact", False)) else "off"
-        },
-        "theme": _cfg_get_theme,
-        "statusbar": lambda rid, params: {
-            "value": _coerce_statusbar(_display_cfg().get("tui_statusbar", "top"))},
-        "focus": lambda rid, params: {
-            "value": "on" if bool(_display_cfg().get("focus_view", False)) else "off",
-            "tool_progress": _load_tool_progress_mode()},
-        "mouse": lambda rid, params: {"value": _display_mouse_tracking(_load_cfg().get("display"))},
-        "mtime": _cfg_get_mtime}
+# key -> getter(params); bind_module rebinds the table's functions onto server.py's globals.
+_CONFIG_GETTERS = {
+    "provider": _cfg_get_provider,
+    "profile": _cfg_get_profile,
+    "project": _cfg_get_project,
+    "full": lambda params: {"config": _load_cfg()},
+    "prompt": lambda params: {"prompt": _load_cfg().get("custom_prompt", "")},
+    "skin": lambda params: {"value": _display_raw().get("skin", "default")},
+    # Normalised like the TUI renders it (frontend falls back to the default for the same inputs).
+    "indicator": lambda params: {
+        "value": _display_word("tui_status_indicator", DEFAULT_INDICATOR_STYLE, INDICATOR_STYLES)},
+    "personality": _cfg_get_personality,
+    "reasoning": _cfg_get_reasoning,
+    "fast": _cfg_get_fast,
+    "busy": lambda params: {"value": _load_busy_input_mode()},
+    "approval_mode": lambda params: {"value": _load_approval_mode()},
+    "approvals.mode": lambda params: {"value": _load_approval_mode()},
+    "details_mode": lambda params: {"value": _display_word("details_mode", "collapsed", _DETAIL_MODES)},
+    "thinking_mode": _cfg_get_thinking_mode,
+    "density": lambda params: {"value": "on" if bool(_display_raw().get("tui_compact", False)) else "off"},
+    "theme": lambda params: {"value": _display_word("tui_theme", "auto", {"auto", "light", "dark"})},
+    "statusbar": lambda params: {"value": _coerce_statusbar(_display_cfg().get("tui_statusbar", "top"))},
+    "focus": lambda params: {"value": "on" if bool(_display_cfg().get("focus_view", False)) else "off",
+                             "tool_progress": _load_tool_progress_mode()},
+    "mouse": lambda params: {"value": _display_mouse_tracking(_load_cfg().get("display"))},
+    "mtime": _cfg_get_mtime}
+# Getters whose failure is a JSON-RPC error of this code (others propagate to dispatch).
+_CONFIG_GET_ERR = {"provider": 5013, "approval_mode": 5001, "approvals.mode": 5001}
 
 
 @method("config.get")
 @_profile_scoped
 def _(rid, params: dict) -> dict:
     key = params.get("key", "")
-    getter = _config_getters().get(key)
+    getter = _CONFIG_GETTERS.get(key)
     if getter is None:
         return _err(rid, 4002, f"unknown config key: {key}")
-    payload = getter(rid, params)
-    if "error" in payload:
-        return payload
-    return _ok(rid, payload)
+    try:
+        return _ok(rid, getter(params))
+    except Exception as e:
+        if key not in _CONFIG_GET_ERR:
+            raise
+        return _err(rid, _CONFIG_GET_ERR[key], str(e))
 
 
 # ── setup readiness
 
-
 def _readiness_profile_scope(params: dict):
     """``(profile, scope)`` for the readiness RPCs' optional ``profile`` param: ``scope`` binds
     that profile's HERMES_HOME + ``.env`` secret scope (ContextVars, so concurrent checks stay
@@ -344,8 +316,7 @@ def _(rid, params: dict) -> dict:
 
         def probe(profile):
             runtime = resolve_runtime_provider(requested=requested)
-            provider_configured = bool(
-                _has_any_provider_configured(strict_profile_scope=bool(profile)))
+            provider_configured = bool(_has_any_provider_configured(strict_profile_scope=bool(profile)))
             scoped = {"profile": profile} if profile else {}
             provider = runtime.get("provider") or "provider"
             source = str(runtime.get("source") or "")
@@ -370,6 +341,14 @@ def _(rid, params: dict) -> dict:
         return _ok(rid, {"ok": False, "error": str(e)})
 
 
+def _safe_client_label(label: str) -> str:
+    """Alnum/._- () only, ≤64 chars, dot-runs and leading dots collapsed (no traversal shapes)."""
+    safe = "".join(ch for ch in label if ch.isalnum() or ch in "._- ()").strip()[:64]
+    while ".." in safe:
+        safe = safe.replace("..", ".")
+    return safe.lstrip(".").strip()
+
+
 @method("diagnostics.share_nous")
 def _(rid, params: dict) -> dict:
     """Upload a redacted debug bundle to Nous-internal diagnostics storage — same collection +
@@ -396,24 +375,17 @@ def _(rid, params: dict) -> dict:
             for label, text in list(extra_files.items())[:4]:
                 if not isinstance(label, str) or not isinstance(text, str):
                     continue
-                safe_label = "".join(ch for ch in label if ch.isalnum() or ch in "._- ()").strip()[:64]
-                # Collapse dot-runs / leading dots so traversal-shaped labels can't survive.
-                while ".." in safe_label:
-                    safe_label = safe_label.replace("..", ".")
-                safe_label = safe_label.lstrip(".").strip()
-                if not safe_label or not text.strip():
-                    continue
-                bundle[f"client/{safe_label}"] = _redact_log_text(text[:524_288])
+                safe_label = _safe_client_label(label)
+                if safe_label and text.strip():
+                    bundle[f"client/{safe_label}"] = _redact_log_text(text[:524_288])
         res = share_to_nous(build_nous_bundle(bundle, redact=True))
         view_url = res.get("viewUrl") or res.get("view_url")
         upload_id = res.get("id")
         if not view_url and not upload_id:
             # An upload the user can't reference is useless to support.
-            return _ok(rid, {"ok": False,
-                             "error": "upload succeeded but returned no view URL or id"})
-        return _ok(rid, {
-            "ok": True, "view_url": view_url, "upload_id": upload_id,
-            "expires_at": res.get("expiresAt") or res.get("expires_at")})
+            return _ok(rid, {"ok": False, "error": "upload succeeded but returned no view URL or id"})
+        return _ok(rid, {"ok": True, "view_url": view_url, "upload_id": upload_id,
+                         "expires_at": res.get("expiresAt") or res.get("expires_at")})
     except Exception as e:
         return _ok(rid, {"ok": False, "error": str(e)})
 
diff --git a/tui_gateway/methods_config_set.py b/tui_gateway/methods_config_set.py
index e9ff07fc87..95b2a882de 100644
--- a/tui_gateway/methods_config_set.py
+++ b/tui_gateway/methods_config_set.py
@@ -44,24 +44,19 @@ def _emit_all_session_info() -> None:
         _emit_session_info(sid, sess)
 
 
-def _toggle_display_bool(rid, key, value, *, cfg_key, on_words, off_words):
-    """Shared body of the on/off/toggle display booleans (``density``, ``battery``)."""
-    raw = _word(value)
-    cur_b = bool(_display_cfg().get(cfg_key, False))
-    if raw in {"", "toggle"}:
-        nv_b = not cur_b
-    elif raw in on_words or raw in off_words:
-        nv_b = raw in on_words
-    else:
-        return _err(rid, 4002, f"unknown {key} value: {value}")
-    _write_config_key(f"display.{cfg_key}", nv_b)
-    return _ok(rid, {"key": key, "value": "on" if nv_b else "off"})
-
-
 def _word(value) -> str:
     return str(value or "").strip().lower()
 
 
+def _raw_word(value) -> str:
+    """Like ``_word`` but only None is blank: falsy non-strings (0, False, []) keep their text."""
+    return ("" if value is None else str(value)).strip().lower()
+
+
+def _kv(rid, key, value, **extra):
+    return _ok(rid, {"key": key, "value": value, **extra})
+
+
 def _cfgset_await_agent(session, rid):
     """Wait for an in-progress agent build; the error envelope if it failed, else None."""
     init_err = _wait_agent(session, rid)
@@ -71,81 +66,85 @@ def _cfgset_await_agent(session, rid):
 
 
 def _cfgset_model_ok(rid, key, value, warning, confirm_required, confirm_message, scope, **extra):
-    return _ok(rid, {"key": key, "value": value, "warning": warning, "confirm_required": confirm_required,
-                     "confirm_message": confirm_message, "scope": scope, **extra})
+    return _kv(rid, key, value, warning=warning, confirm_required=confirm_required,
+               confirm_message=confirm_message, scope=scope, **extra)
+
+
+def _cfgset_guarded(fn):
+    """Setter whose uncaught exception becomes ``_err(rid, 5001, str(e))``."""
+    def setter(rid, params, key, value, session):
+        try:
+            return fn(rid, params, key, value, session)
+        except Exception as e:
+            return _err(rid, 5001, str(e))
+    return setter
 
 
 # ── per-key handlers
 
+@_cfgset_guarded
 def _set_model(rid, params, key, value, session):
     """Live/deferred model switch; see _apply_model_switch and _apply_pending_model_switch."""
-    try:
-        if not value:
-            return _err(rid, 4002, "model value required")
-        confirmed = bool(params.get("confirm_expensive_model", False))
-        if session:
-            from hermes_cli.model_switch import parse_model_switch_args
-            sid = params.get("session_id", "")
-            # No live swap while a turn streams (agent.switch_model() mutates fields the worker
-            # thread reads every iteration): stash the pick for the NEXT turn start.
-            if session.get("running"):
-                parsed = parse_model_switch_args(value)
-                try:
-                    pending_model = parsed.model_input
-                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.
-                if not confirmed:
-                    pending_warning = _pending_switch_selection_warning(pending_model, pending_provider)
-                    if pending_warning is not None:
-                        # Nothing stashed; the client re-sends with confirm_expensive_model.
-                        # `confirm_message` is canonical, `warning` its legacy alias.
-                        return _cfgset_model_ok(
-                            rid, key, pending_model, pending_warning, True, pending_warning, "session", deferred=False
-                        )
-                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)
-            explicit_provider = parsed_flags.explicit_provider
-            failed_agent_init = session.get("agent") is None and session.get("agent_error") is not None
-            failed_ready = session.get("agent_ready") if failed_agent_init else None
-            if failed_agent_init:
-                if failed_ready is None:
-                    return _err(rid, 5032, session.get("agent_error") or "agent initialization failed")
-                if not failed_ready.wait(timeout=30.0):
-                    return _err(rid, 5032, "agent initialization timed out")
-            failed_agent_init = (
-                failed_agent_init and session.get("agent") is None and session.get("agent_error") is not None
-                and session.get("agent_ready") is failed_ready and failed_ready.is_set())
-            if session.get("agent") is None and not explicit_provider.strip() and not failed_agent_init:
-                _start_agent_build(sid, session)
-                if init_err := _cfgset_await_agent(session, rid):
-                    return init_err
+    if not value:
+        return _err(rid, 4002, "model value required")
+    confirmed = bool(params.get("confirm_expensive_model", False))
+    if session:
+        from hermes_cli.model_switch import parse_model_switch_args
+        sid = params.get("session_id", "")
+        # No live swap while a turn streams (agent.switch_model() mutates fields the worker
+        # thread reads every iteration): stash the pick for the NEXT turn start.
+        if session.get("running"):
+            parsed = parse_model_switch_args(value)
+            try:
+                pending_model = parsed.model_input
+            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.
+            # `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)
+            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)
+        explicit_provider = parsed_flags.explicit_provider
+        failed_agent_init = session.get("agent") is None and session.get("agent_error") is not None
+        failed_ready = session.get("agent_ready") if failed_agent_init else None
+        if failed_agent_init:
+            if failed_ready is None:
+                return _err(rid, 5032, session.get("agent_error") or "agent initialization failed")
+            if not failed_ready.wait(timeout=30.0):
+                return _err(rid, 5032, "agent initialization timed out")
+        failed_agent_init = (
+            failed_agent_init and session.get("agent") is None and session.get("agent_error") is not None
+            and session.get("agent_ready") is failed_ready and failed_ready.is_set())
+        if session.get("agent") is None and not explicit_provider.strip() and not failed_agent_init:
+            _start_agent_build(sid, session)
+            if init_err := _cfgset_await_agent(session, rid):
+                return init_err
+        with _session_profile_runtime_scope(session):
+            result = _apply_model_switch(sid, session, value, confirm_expensive_model=confirmed,
+                                         parsed_flags=parsed_flags)
+        if failed_agent_init and not result.get("confirm_required"):
+            _restart_completed_failed_agent_build(sid, session, failed_ready)
+            if init_err := _cfgset_await_agent(session, rid):
+                return init_err
             with _session_profile_runtime_scope(session):
-                result = _apply_model_switch(
-                    sid, session, value, confirm_expensive_model=confirmed, parsed_flags=parsed_flags
-                )
-            if failed_agent_init and not result.get("confirm_required"):
-                _restart_completed_failed_agent_build(sid, session, failed_ready)
-                if init_err := _cfgset_await_agent(session, rid):
-                    return init_err
-                with _session_profile_runtime_scope(session):
-                    _persist_live_session_runtime(session)
-        else:
-            result = _apply_model_switch("", {"agent": None}, value, confirm_expensive_model=confirmed)
-        return _cfgset_model_ok(
-            rid, key, result["value"], result["warning"], result.get("confirm_required", False),
-            result.get("confirm_message", ""), result.get("scope", "session"))
-    except Exception as e:
-        return _err(rid, 5001, str(e))
+                _persist_live_session_runtime(session)
+    else:
+        result = _apply_model_switch("", {"agent": None}, value, confirm_expensive_model=confirmed)
+    return _cfgset_model_ok(
+        rid, key, result["value"], result["warning"], result.get("confirm_required", False),
+        result.get("confirm_message", ""), result.get("scope", "session"))
 
 
 _FAST_WORDS = {"fast": "fast", "on": "fast", "normal": "normal", "off": "normal",
@@ -162,10 +161,9 @@ def _set_fast(rid, params, key, value, session):
         current_tier = session["create_service_tier_override"] or None
     else:
         current_tier = _load_service_tier()
-    current_fast = current_tier == "priority"
     if raw == "status":
-        return _ok(rid, {"key": key, "value": {"priority": "fast", None: "normal"}.get(current_tier, current_tier)})
-    toggled = ("normal" if current_fast else "fast") if raw in {"", "toggle"} else None
+        return _kv(rid, key, {"priority": "fast", None: "normal"}.get(current_tier, current_tier))
+    toggled = ("normal" if current_tier == "priority" else "fast") if raw in {"", "toggle"} else None
     nv = _FAST_WORDS.get(raw, toggled)
     if nv is None:
         return _err(rid, 4002, f"unknown fast mode: {value}")
@@ -201,17 +199,17 @@ def _set_fast(rid, params, key, value, session):
         agent.request_overrides = current_overrides
         _persist_live_session_runtime(session)
         _emit_session_info(params.get("session_id", ""), session)
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
 def _set_busy(rid, params, key, value, session):
     raw = _word(value)
     if raw in {"", "status"}:
-        return _ok(rid, {"key": key, "value": _load_busy_input_mode()})
+        return _kv(rid, key, _load_busy_input_mode())
     if raw not in {"queue", "steer", "interrupt"}:
         return _err(rid, 4002, f"unknown busy mode: {value}")
     _write_config_key("display.busy_input_mode", raw)
-    return _ok(rid, {"key": key, "value": raw})
+    return _kv(rid, key, raw)
 
 
 def _set_verbose(rid, params, key, value, session):
@@ -229,7 +227,7 @@ def _set_verbose(rid, params, key, value, session):
         session["tool_progress_mode"] = nv
         if session.get("agent") is not None:
             session["agent"].verbose_logging = nv == "verbose"
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
 def _set_focus(rid, params, key, value, session):
@@ -242,15 +240,14 @@ def _set_focus(rid, params, key, value, session):
     if action == "usage":
         return _err(rid, 4002, f"unknown focus value: {value} (use on|off|status)")
     if action == "status" or target is None:
-        return _ok(rid, {"key": key, "value": "on" if cur_focus else "off", "tool_progress": _load_tool_progress_mode()})
+        return _kv(rid, key, "on" if cur_focus else "off", tool_progress=_load_tool_progress_mode())
     if target:
         saved = (cur_focus and d_f.get("focus_saved_tool_progress")) or _load_tool_progress_mode()
         _write_config_key("display.focus_saved_tool_progress", normalize_tool_progress_mode(saved))
-        _write_config_key("display.tool_progress", FOCUS_TOOL_PROGRESS_MODE)
         effective = FOCUS_TOOL_PROGRESS_MODE
     else:
         effective = normalize_tool_progress_mode(d_f.get("focus_saved_tool_progress") or "all")
-        _write_config_key("display.tool_progress", effective)
+    _write_config_key("display.tool_progress", effective)
     _write_config_key("display.focus_view", bool(target))
     if session:
         session["focus_view"] = bool(target)
@@ -258,7 +255,7 @@ def _set_focus(rid, params, key, value, session):
         if session.get("agent") is not None:
             with contextlib.suppress(Exception):
                 session["agent"].tool_progress_mode = effective
-    return _ok(rid, {"key": key, "value": "on" if target else "off", "tool_progress": effective})
+    return _kv(rid, key, "on" if target else "off", tool_progress=effective)
 
 
 def _set_approval_mode(rid, params, key, value, session):
@@ -267,42 +264,40 @@ def _set_approval_mode(rid, params, key, value, session):
         return _err(rid, 4002, f"unknown approval mode: {value}; pick one of manual|smart|off")
     _write_config_key("approvals.mode", raw)
     _emit_all_session_info()
-    return _ok(rid, {"key": "approvals.mode", "value": raw})
+    return _kv(rid, "approvals.mode", raw)
 
 
+@_cfgset_guarded
 def _set_yolo(rid, params, key, value, session):
     # 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")
-    try:
-        from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled
-        raw = _word(value)
+    from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled
+    raw = _word(value)
 
-        def _resolve_toggle(current: bool) -> bool:
-            return _BOOL_WORDS.get(raw, not current)
-        if scope == "global":
-            from tools.approval import _normalize_approval_mode
-            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")
-            _emit_all_session_info()  # reflect the flip in every live indicator
-            return _ok(rid, {"key": key, "value": "1" if enable else "0", "scope": "global"})
-        if session:
-            skey = session["session_key"]
-            enable = _resolve_toggle(is_session_yolo_enabled(skey))
-            (enable_session_yolo if enable else disable_session_yolo)(skey)
-            _emit_session_info(params.get("session_id", ""), session)
+    def _resolve_toggle(current: bool) -> bool:
+        return _BOOL_WORDS.get(raw, not current)
+    if scope == "global":
+        from tools.approval import _normalize_approval_mode
+        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")
+        _emit_all_session_info()  # reflect the flip in every live indicator
+        return _kv(rid, key, "1" if enable else "0", scope="global")
+    if session:
+        skey = session["session_key"]
+        enable = _resolve_toggle(is_session_yolo_enabled(skey))
+        (enable_session_yolo if enable else disable_session_yolo)(skey)
+        _emit_session_info(params.get("session_id", ""), session)
+    else:
+        enable = _resolve_toggle(is_truthy_value(os.environ.get("HERMES_YOLO_MODE")))
+        if enable:
+            os.environ["HERMES_YOLO_MODE"] = "1"
         else:
-            enable = _resolve_toggle(is_truthy_value(os.environ.get("HERMES_YOLO_MODE")))
-            if enable:
-                os.environ["HERMES_YOLO_MODE"] = "1"
-            else:
-                os.environ.pop("HERMES_YOLO_MODE", None)
-        return _ok(rid, {"key": key, "value": "1" if enable else "0", "scope": "session"})
-    except Exception as e:
-        return _err(rid, 5001, str(e))
+            os.environ.pop("HERMES_YOLO_MODE", None)
+    return _kv(rid, key, "1" if enable else "0", scope="session")
 
 
 # /reasoning display words: (accepted inputs, reported value, display field, sections.thinking,
@@ -314,35 +309,33 @@ _REASONING_DISPLAY_WORDS = (
     ({"clamp", "collapse", "short"}, "clamp", {"reasoning_full": False}, "collapsed", None))
 
 
+@_cfgset_guarded
 def _set_reasoning(rid, params, key, value, session):
-    try:
-        from hermes_constants import parse_reasoning_effort
-        arg = _word(value)
-        scope = _word(params.get("scope"))
-        for words, reported, fields, thinking, show in _REASONING_DISPLAY_WORDS:
-            if arg in words:
-                _write_display_sections(sections={"thinking": thinking}, **fields)
-                if show is not None and session:
-                    session["show_reasoning"] = show
-                return _ok(rid, {"key": key, "value": reported})
-        parsed = parse_reasoning_effort(arg)
-        if parsed is None:
-            return _err(rid, 4002, f"unknown reasoning value: {value}")
-        if scope == "global" or session is None:
-            _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.
-            session["create_reasoning_override"] = parsed
-        if session and session.get("agent") is not None:
-            session["agent"].reasoning_config = parsed
-            _persist_live_session_runtime(session)
-            _emit_session_info(params.get("session_id", ""), session)
-        return _ok(rid, {"key": key, "value": arg})
-    except Exception as e:
-        return _err(rid, 5001, str(e))
+    from hermes_constants import parse_reasoning_effort
+    arg = _word(value)
+    scope = _word(params.get("scope"))
+    for words, reported, fields, thinking, show in _REASONING_DISPLAY_WORDS:
+        if arg in words:
+            _write_display_sections(sections={"thinking": thinking}, **fields)
+            if show is not None and session:
+                session["show_reasoning"] = show
+            return _kv(rid, key, reported)
+    parsed = parse_reasoning_effort(arg)
+    if parsed is None:
+        return _err(rid, 4002, f"unknown reasoning value: {value}")
+    if scope == "global" or session is None:
+        _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.
+        session["create_reasoning_override"] = parsed
+    if session and session.get("agent") is not None:
+        session["agent"].reasoning_config = parsed
+        _persist_live_session_runtime(session)
+        _emit_session_info(params.get("session_id", ""), session)
+    return _kv(rid, key, arg)
 
 
 def _set_details_mode(rid, params, key, value, session):
@@ -350,7 +343,7 @@ def _set_details_mode(rid, params, key, value, session):
     if nv not in _DETAIL_MODES:
         return _err(rid, 4002, f"unknown details_mode: {value}")
     _write_display_sections(sections={section: nv for section in _DETAIL_SECTION_NAMES}, details_mode=nv)
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
 def _set_details_section(rid, params, key, value, session):
@@ -366,7 +359,7 @@ def _set_details_section(rid, params, key, value, session):
         return _err(rid, 4002, f"unknown details_mode: {value}")
     else:
         _write_display_sections(sections={section: nv})
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
 def _set_thinking_mode(rid, params, key, value, session):
@@ -376,17 +369,27 @@ def _set_thinking_mode(rid, params, key, value, session):
     _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")
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
-def _set_density(rid, params, key, value, session):
-    return _toggle_display_bool(rid, key, value, cfg_key="tui_compact", on_words={"on"}, off_words={"off"})
+# on/off/toggle display booleans: key -> (display field, on words, off words).
+_DISPLAY_BOOLS = {
+    "density": ("tui_compact", {"on"}, {"off"}),
+    "battery": ("battery", {"on", "true", "yes"}, {"off", "false", "no"})}
 
 
-def _set_battery(rid, params, key, value, session):
-    return _toggle_display_bool(
-        rid, key, value, cfg_key="battery", on_words={"on", "true", "yes"}, off_words={"off", "false", "no"}
-    )
+def _set_display_bool(rid, params, key, value, session):
+    cfg_key, on_words, off_words = _DISPLAY_BOOLS[key]
+    raw = _word(value)
+    cur_b = bool(_display_cfg().get(cfg_key, False))
+    if raw in {"", "toggle"}:
+        nv_b = not cur_b
+    elif raw in on_words or raw in off_words:
+        nv_b = raw in on_words
+    else:
+        return _err(rid, 4002, f"unknown {key} value: {value}")
+    _write_config_key(f"display.{cfg_key}", nv_b)
+    return _kv(rid, key, "on" if nv_b else "off")
 
 
 def _set_theme(rid, params, key, value, session):
@@ -395,7 +398,7 @@ def _set_theme(rid, params, key, value, session):
     if raw not in {"auto", "light", "dark"}:
         return _err(rid, 4002, f"unknown theme value: {value} (use auto|light|dark)")
     _write_config_key("display.tui_theme", raw)
-    return _ok(rid, {"key": key, "value": raw})
+    return _kv(rid, key, raw)
 
 
 def _set_statusbar(rid, params, key, value, session):
@@ -408,13 +411,11 @@ def _set_statusbar(rid, params, key, value, session):
     else:
         return _err(rid, 4002, f"unknown statusbar value: {value}")
     _write_config_key("display.tui_statusbar", nv)
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
 def _set_mouse(rid, params, key, value, session):
-    # Explicit None check so falsy non-string inputs (0, False) reach the alias map as
-    # themselves (-> 'off') instead of toggling.
-    raw = ("" if value is None else str(value)).strip().lower()
+    raw = _raw_word(value)  # 0/False reach the alias map as themselves (-> 'off'), not toggle
     current = _display_mouse_tracking(_display_cfg())
     if raw in {"", "toggle"}:
         nv = "all" if current == "off" else "off"
@@ -423,16 +424,15 @@ def _set_mouse(rid, params, key, value, session):
     else:
         return _err(rid, 4002, f"unknown mouse value: {value}")
     _write_config_key("display.mouse_tracking", nv)
-    return _ok(rid, {"key": key, "value": nv})
+    return _kv(rid, key, nv)
 
 
 def _set_indicator(rid, params, key, value, session):
-    # Explicit None check so falsy non-string inputs (0, False, []) surface in the error message.
-    raw = ("" if value is None else str(value)).strip().lower()
+    raw = _raw_word(value)  # 0/False/[] surface in the error message
     if raw not in INDICATOR_STYLES:
         return _err(rid, 4002, f"unknown indicator: {raw!r}; pick one of {'|'.join(INDICATOR_STYLES)}")
     _write_config_key("display.tui_status_indicator", raw)
-    return _ok(rid, {"key": key, "value": raw})
+    return _kv(rid, key, raw)
 
 
 def _set_cwd(rid, params, key, value, session):
@@ -444,41 +444,39 @@ def _set_cwd(rid, params, key, value, session):
         return _err(rid, 4002, f"working directory does not exist: {raw}")
     _write_config_key("terminal.cwd", cwd)
     os.environ["TERMINAL_CWD"] = cwd
-    return _ok(rid, {"key": "terminal.cwd", "value": cwd, "cwd": cwd, "branch": _git_branch_for_cwd(cwd)})
+    return _kv(rid, "terminal.cwd", cwd, cwd=cwd, branch=_git_branch_for_cwd(cwd))
 
 
+@_cfgset_guarded
 def _set_prompt_like(rid, params, key, value, session):
-    try:
-        cfg = _load_cfg_raw()  # write-back round-trip ("prompt" saves cfg)
-        resp = {"key": key, "value": value}
-        if key == "prompt":
-            if value == "clear":
-                cfg.pop("custom_prompt", None)
-                resp["value"] = ""
-            else:
-                cfg["custom_prompt"] = value
-            _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.
-            from hermes_cli.personality import persist_personality
-            persist_personality(pname)
-            resp["value"] = str(value or "none")
-            history_reset, info = _apply_personality_to_session(params.get("session_id", ""), session, new_prompt, pname)
-            resp["history_reset"] = history_reset
-            if info is not None:
-                resp["info"] = info
+    cfg = _load_cfg_raw()  # write-back round-trip ("prompt" saves cfg)
+    resp = {"key": key, "value": value}
+    if key == "prompt":
+        if value == "clear":
+            cfg.pop("custom_prompt", None)
+            resp["value"] = ""
         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.
-                _broadcast_global_event("skin.changed", resolve_skin())
-                _note_skin_broadcast()
-        return _ok(rid, resp)
-    except Exception as e:
-        return _err(rid, 5001, str(e))
+            cfg["custom_prompt"] = value
+        _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.
+        from hermes_cli.personality import persist_personality
+        persist_personality(pname)
+        resp["value"] = str(value or "none")
+        history_reset, info = _apply_personality_to_session(params.get("session_id", ""), session, new_prompt, pname)
+        resp["history_reset"] = history_reset
+        if info is not None:
+            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.
+            _broadcast_global_event("skin.changed", resolve_skin())
+            _note_skin_broadcast()
+    return _ok(rid, resp)
 
 
 def _set_display_toggle(rid, params, key, value, session):
@@ -486,7 +484,7 @@ def _set_display_toggle(rid, params, key, value, session):
     if on is None:
         return _err(rid, 4002, f"{key} takes true or false")
     _write_config_key(key, on)
-    return _ok(rid, {"key": key, "value": on})
+    return _kv(rid, key, on)
 
 
 # ── dispatch
@@ -495,8 +493,8 @@ _CONFIG_SETTERS = {
     "model": _set_model, "fast": _set_fast, "busy": _set_busy, "verbose": _set_verbose, "focus": _set_focus,
     "approval_mode": _set_approval_mode, "approvals.mode": _set_approval_mode, "yolo": _set_yolo,
     "reasoning": _set_reasoning, "details_mode": _set_details_mode, "thinking_mode": _set_thinking_mode,
-    "density": _set_density, "battery": _set_battery, "theme": _set_theme, "statusbar": _set_statusbar,
-    "mouse": _set_mouse, "indicator": _set_indicator,
+    "density": _set_display_bool, "battery": _set_display_bool, "theme": _set_theme,
+    "statusbar": _set_statusbar, "mouse": _set_mouse, "indicator": _set_indicator,
     "cwd": _set_cwd, "terminal.cwd": _set_cwd, "workdir": _set_cwd,
     "prompt": _set_prompt_like, "personality": _set_prompt_like, "skin": _set_prompt_like}
 
diff --git a/tui_gateway/methods_profiles.py b/tui_gateway/methods_profiles.py
index 9792e3c5eb..990b620d33 100644
--- a/tui_gateway/methods_profiles.py
+++ b/tui_gateway/methods_profiles.py
@@ -39,11 +39,6 @@ def _pin_profile_model(profile_dir, provider, model) -> None:
     _lazy("hermes_cli.web_routers.profiles", "_write_profile_model")(profile_dir, provider, model)
 
 
-def _launch_mcp_catalog() -> dict:
-    mcp = (_lazy("hermes_cli.config", "load_config_readonly")() or {}).get("mcp_servers")
-    return mcp if isinstance(mcp, dict) else {}
-
-
 def _try(fn, default):
     """``fn()`` or ``default`` on any exception — best-effort sections must never fail each other."""
     try:
@@ -111,15 +106,6 @@ def _latest_message_preview(db, session_id):
     return text[:80] + "..." if len(text) > 80 else text
 
 
-def _open_profile_session_db_readonly(profile_path):
-    """Read-only attach for roster previews, or None (a writable ``SessionDB()`` waits up to 20s
-    for the write lock + runs DDL and stalled the 5s roster poll)."""
-    db_path = Path(profile_path) / "state.db"
-    if not _try(db_path.exists, False):
-        return None
-    return _try(lambda: _lazy("hermes_state", "SessionDB")(db_path=db_path, read_only=True), None)
-
-
 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."""
@@ -145,13 +131,9 @@ def _canonical_session_row(db, profile_path):
     """Summary of the profile's canonical "Bot Chat" row (identity is the NAME), or None.
     Lineages via ``get_compression_tip`` (NOT the resume walker's unmarked-child fallback);
     worker sources count as absent. ``id`` is the registry row, ``resolved_id`` the live tip."""
-    if db is None:
-        return None
     try:
         row = db.get_session_by_title("Bot Chat")
-        if not row:
-            return None
-        session_id = str(row.get("id") or "").strip()
+        session_id = str((row or {}).get("id") or "").strip()
         if not session_id or _denied_source(row):
             return None
         # Archived = retired (absent), except accidental reaper archives: resurrect those.
@@ -173,8 +155,6 @@ 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)."""
-    if db is None:
-        return None, None
     try:
         human = worker = None
         for s in db.list_sessions_rich(source=None, limit=20, order_by_last_active=True, compact_rows=True):
@@ -184,29 +164,32 @@ def _latest_profile_session_rows(db):
                 if worker is None:
                     src = (s.get("source") or "").strip().lower()
                     worker = {"id": s["id"], "source": src, "title": title, "last_active": last_active}
-                continue
-            if human is not None:
-                continue
-            # Rosters want "where the conversation IS": prefer the newest text.
-            human = {
-                "id": s["id"], "title": title,
-                "preview": _latest_message_preview(db, s["id"]) or s.get("preview") or "",
-                "started_at": s.get("started_at") or 0, "last_active": last_active,
-                "message_count": s.get("message_count") or 0}
-            if worker is not None:
-                break
+            elif human is None:
+                # Rosters want "where the conversation IS": prefer the newest text.
+                human = {
+                    "id": s["id"], "title": title,
+                    "preview": _latest_message_preview(db, s["id"]) or s.get("preview") or "",
+                    "started_at": s.get("started_at") or 0, "last_active": last_active,
+                    "message_count": s.get("message_count") or 0}
+                if worker is not None:
+                    break
         return human, worker
     except Exception:
         return None, None
 
 
 def _profile_session_fields(row, profile_path):
-    """Attach last_session / worker_session / canonical_session to a roster row."""
-    db = _open_profile_session_db_readonly(profile_path)
+    """Attach last_session / worker_session / canonical_session to a roster row. The DB is a
+    read-only attach (a writable ``SessionDB()`` waits up to 20s for the write lock + runs DDL
+    and stalled the 5s roster poll); no DB -> every field None."""
+    db_path = Path(profile_path) / "state.db"
+    db = None
+    if _try(db_path.exists, False):
+        db = _try(lambda: _lazy("hermes_state", "SessionDB")(db_path=db_path, read_only=True), None)
     try:
-        row["last_session"], row["worker_session"] = _latest_profile_session_rows(db)
+        row["last_session"], row["worker_session"] = _latest_profile_session_rows(db) if db else (None, None)
         # Resolved server-side on every listing so no client carries a session pointer.
-        row["canonical_session"] = _canonical_session_row(db, profile_path)
+        row["canonical_session"] = _canonical_session_row(db, profile_path) if db else None
     finally:
         if db is not None:
             _best_effort(db.close)
@@ -250,12 +233,6 @@ def _(rid, params: dict) -> dict:
     return _ok(rid, {"profiles": out, "bot_mode_protocol": True})
 
 
-def _has_real_env_content(env_path) -> bool:
-    """True when .env has any non-comment, non-blank line."""
-    lines = env_path.read_text(encoding="utf-8", errors="replace").splitlines()
-    return any(s and not s.startswith("#") for s in map(str.strip, lines))
-
-
 def _copy_secret_file(src, dst, wanted: bool) -> bool:
     """Copy ``src`` -> ``dst`` (0600) when ``src`` exists and ``wanted``; True if copied."""
     if not (src.is_file() and wanted):
@@ -269,9 +246,11 @@ def _copy_secret_file(src, dst, wanted: bool) -> bool:
 
 def _mirror_env(path, launch_home) -> bool:
     """Copy the launch .env only over the seeded comment-only stub (never a clone's secrets)."""
+    def has_content(env_path) -> bool:
+        lines = env_path.read_text(encoding="utf-8", errors="replace").splitlines()
+        return any(s and not s.startswith("#") for s in map(str.strip, lines))
     src, dst = launch_home / ".env", path / ".env"
-    return _copy_secret_file(
-        src, dst, _has_real_env_content(src) and not _try(lambda: _has_real_env_content(dst), False))
+    return _copy_secret_file(src, dst, has_content(src) and not _try(lambda: has_content(dst), False))
 
 
 def _mirror_auth(path, launch_home) -> bool:
@@ -408,22 +387,10 @@ def _describe_toolsets(cfg):
     return toolsets_out, pinned_set
 
 
-def _describe_mcp_servers(cfg):
-    """``[{name, enabled, transport}]`` for the profile's ``mcp_servers`` (best-effort)."""
-    mcp_cfg = cfg.get("mcp_servers")
-    if not isinstance(mcp_cfg, dict):
-        return []
-    return _try(lambda: [
-        {"name": str(srv_name), "enabled": not is_truthy_value(entry.get("disabled", False)),
-         "transport": str(entry.get("transport") or "http") if entry.get("url") else "stdio"}
-        for srv_name in sorted(mcp_cfg.keys()) for entry in (mcp_cfg[srv_name],)
-        if isinstance(entry, dict)
-    ], [])
-
-
 @_profile_handler("profiles.describe", 5063)
 def _(rid, params: dict) -> dict:
-    """Editor snapshot; installed skills are enabled unless in ``skills.disabled``."""
+    """Editor snapshot; installed skills are enabled unless in ``skills.disabled``; ``mcp_servers``
+    is ``[{name, enabled, transport}]`` (best-effort)."""
     name, profile_dir, err = _resolve_profile(rid, params)
     if err is not None:
         return err
@@ -439,7 +406,13 @@ def _(rid, params: dict) -> dict:
         toolsets_out, pinned_set = _describe_toolsets(cfg)
         soul_path = profile_dir / "SOUL.md"
         soul = _try(lambda: soul_path.read_text(encoding="utf-8", errors="replace") if soul_path.is_file() else "", "")
-        mcp_out = _describe_mcp_servers(cfg)
+        mcp_cfg = cfg.get("mcp_servers")
+        mcp_out = _try(lambda: [
+            {"name": str(srv_name), "enabled": not is_truthy_value(entry.get("disabled", False)),
+             "transport": str(entry.get("transport") or "http") if entry.get("url") else "stdio"}
+            for srv_name in sorted(mcp_cfg.keys()) for entry in (mcp_cfg[srv_name],)
+            if isinstance(entry, dict)
+        ], []) if isinstance(mcp_cfg, dict) else []
         model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {}
         meta = _try(lambda: _lazy("hermes_cli.profiles", "read_profile_meta")(profile_dir), {})
         return _ok(rid, {
@@ -513,31 +486,6 @@ def _configure_model(profile_dir, params, applied):
     return confirm_message
 
 
-def _configure_cfg_sections(profile_dir, params, applied) -> None:
-    """Apply ``disabled_skills`` / ``enabled_toolsets`` / ``enabled_mcp_servers`` (replace
-    semantics; empty toolsets clears the pin). An undefined MCP server is copied from the LAUNCH
-    catalog (unknown names skipped); credentials stay in .env/auth."""
-    want_mcp = isinstance(params.get("enabled_mcp_servers"), list)
-    # Launch catalog read BEFORE the home override flips config resolution.
-    launch_mcp = _try(_launch_mcp_catalog, {}) if want_mcp else {}
-    with _hermes_home_scope(profile_dir):
-        from hermes_cli.config import load_config, save_config
-        cfg = load_config() or {}
-        if isinstance(params.get("disabled_skills"), list):
-            try:
-                from hermes_cli.skills_config import save_disabled_skills
-                save_disabled_skills(cfg, _clean_names(params["disabled_skills"]))
-                applied["skills"] = True
-                cfg = load_config() or {}
-            except Exception:
-                applied["skills"] = False
-        if isinstance(params.get("enabled_toolsets"), list):
-            applied["toolsets"] = _best_effort(lambda: _save_toolset_pin(cfg, params["enabled_toolsets"], save_config))
-        if want_mcp:
-            applied["mcp_servers"] = _best_effort(lambda: _save_mcp_toggles(
-                load_config() or {}, params["enabled_mcp_servers"], launch_mcp, save_config))
-
-
 def _clean_names(values) -> set:
     return {str(v).strip() for v in values if str(v).strip()}
 
@@ -570,6 +518,34 @@ def _save_mcp_toggles(cfg, enabled, launch_mcp, save_config) -> None:
     save_config(cfg)
 
 
+def _configure_cfg_sections(profile_dir, params, applied) -> None:
+    """Apply ``disabled_skills`` / ``enabled_toolsets`` / ``enabled_mcp_servers`` (replace
+    semantics; empty toolsets clears the pin). An undefined MCP server is copied from the LAUNCH
+    catalog (unknown names skipped); credentials stay in .env/auth."""
+    want_mcp = isinstance(params.get("enabled_mcp_servers"), list)
+    # 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"), {})
+        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
+        cfg = load_config() or {}
+        if isinstance(params.get("disabled_skills"), list):
+            try:
+                from hermes_cli.skills_config import save_disabled_skills
+                save_disabled_skills(cfg, _clean_names(params["disabled_skills"]))
+                applied["skills"] = True
+                cfg = load_config() or {}
+            except Exception:
+                applied["skills"] = False
+        if isinstance(params.get("enabled_toolsets"), list):
+            applied["toolsets"] = _best_effort(lambda: _save_toolset_pin(cfg, params["enabled_toolsets"], save_config))
+        if want_mcp:
+            applied["mcp_servers"] = _best_effort(lambda: _save_mcp_toggles(
+                load_config() or {}, params["enabled_mcp_servers"], launch_mcp, save_config))
+
+
 @_profile_handler("profiles.configure", 5064)
 def _(rid, params: dict) -> dict:
     """Editor Save: ``name`` plus any of ``ui_meta`` (+ ``ui_meta_expected_revisions``), ``soul``,
@@ -598,15 +574,6 @@ def _(rid, params: dict) -> dict:
     return _ok(rid, result)
 
 
-def _sniff_asset_ext(blob):
-    """Extension for a PNG/JPEG/WebP blob by magic bytes (never trust declared mime), or None."""
-    if blob[:8] == b"\x89PNG\r\n\x1a\n":
-        return "png"
-    if blob[:3] == b"\xff\xd8\xff":
-        return "jpg"
-    return "webp" if blob[:4] == b"RIFF" and blob[8:12] == b"WEBP" else None
-
-
 def _unlink_asset_files(assets_dir, asset) -> int:
     """Delete every ``<asset>.<ext>`` in ``assets_dir``; returns how many existed."""
     present = [t for t in (assets_dir / f"{asset}.{ext}" for ext in _ASSET_EXTS) if t.is_file()]
@@ -618,7 +585,8 @@ 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) or ``clear: true``."""
+    ``data`` (data URL or base64; PNG/JPEG/WebP ≤2MB, sniffed by magic bytes — never the declared
+    mime) 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")
@@ -643,8 +611,13 @@ def _(rid, params: dict) -> dict:
         return _err(rid, 4068, "data is not valid base64")
     if len(blob) > 2_000_000:
         return _err(rid, 4069, f"asset too large ({len(blob)} bytes; max 2MB)")
-    ext = _sniff_asset_ext(blob)
-    if ext is None:
+    if blob[:8] == b"\x89PNG\r\n\x1a\n":
+        ext = "png"
+    elif blob[:3] == b"\xff\xd8\xff":
+        ext = "jpg"
+    elif blob[:4] == b"RIFF" and blob[8:12] == b"WEBP":
+        ext = "webp"
+    else:
         return _err(rid, 4070, "unsupported image format (PNG/JPEG/WebP only)")
     assets_dir.mkdir(parents=True, exist_ok=True)
     _unlink_asset_files(assets_dir, asset)  # one canonical file per asset
diff --git a/tui_gateway/methods_voice.py b/tui_gateway/methods_voice.py
index 52cf5b87e2..832b1123c5 100644
--- a/tui_gateway/methods_voice.py
+++ b/tui_gateway/methods_voice.py
@@ -14,7 +14,8 @@ _registry = HandlerRegistry()
 method = _registry.method
 
 
-# ── Voice state ──────────────────────────────────────────────────────────
+# ── Voice state: HERMES_VOICE / HERMES_VOICE_TTS are runtime-only env flags (never config.yaml)
+# so a prior session can't auto-start REC.
 
 _voice_sid_lock = threading.Lock()
 _voice_event_sid: str = ""
@@ -41,12 +42,10 @@ def _resume_voice_wake() -> None:
 
 
 def _voice_mode_enabled() -> bool:
-    """Runtime-only flag (env, never config.yaml) so a prior session can't auto-start REC."""
     return os.environ.get("HERMES_VOICE", "").strip() == "1"
 
 
 def _voice_tts_enabled() -> bool:
-    """Whether agent replies are spoken back via TTS (runtime only)."""
     return os.environ.get("HERMES_VOICE_TTS", "").strip() == "1"
 
 
@@ -75,11 +74,15 @@ def _tts_lease_async(lease: str, active: bool) -> None:
     threading.Thread(target=_run, name=f"tts-lease-{lease}", daemon=True).start()
 
 
+def _running_sessions() -> list:
+    with _sessions_lock:
+        return [s for s in _sessions.values() if s.get("running")]
+
+
 def _any_session_running() -> bool:
     """Voice busy-probe: silent captures during a long turn don't count toward the no-speech limit."""
     try:
-        with _sessions_lock:
-            return any(s.get("running") for s in _sessions.values())
+        return bool(_running_sessions())
     except Exception:
         return False
 
@@ -104,8 +107,7 @@ def _tts_stream_begin() -> Optional[queue.Queue]:
     _tts_stream_stop()
     text_queue: queue.Queue = queue.Queue()
     stop, done = threading.Event(), threading.Event()
-    threading.Thread(target=stream_tts_to_speaker, args=(text_queue, stop, done),
-                     daemon=True).start()
+    threading.Thread(target=stream_tts_to_speaker, args=(text_queue, stop, done), daemon=True).start()
     global _tts_stream_state
     with _tts_stream_lock:
         _tts_stream_state = {"stop": stop, "done": done}
@@ -237,8 +239,9 @@ def _cut_all_tts() -> None:
 
 
 def _fd_trip(phase: str) -> None:
-    """Listener tripped: latch the interruption, cut TTS, and during generation also
-    interrupt every running turn (the ``agent.interrupt()`` seam ``session.interrupt`` uses)."""
+    """Listener tripped: latch the interruption, cut TTS FIRST (so a stale reply can never
+    speak), and during generation also interrupt every running turn (the ``agent.interrupt()``
+    seam ``session.interrupt`` uses)."""
     from tools.tts_streaming import mark_speech_interrupted
     mark_speech_interrupted()
     if phase == "playback":
@@ -247,12 +250,9 @@ def _fd_trip(phase: str) -> None:
     else:
         logger.debug("full-duplex listener tripped during generation — "
                      "interrupting running turn(s)")
-        # Cut pending TTS FIRST so the stale reply can never speak.
         _cut_all_tts()
         try:
-            with _sessions_lock:
-                running = [s for s in _sessions.values() if s.get("running")]
-            for s in running:
+            for s in _running_sessions():
                 agent = s.get("agent")
                 if agent is not None and hasattr(agent, "interrupt"):
                     with contextlib.suppress(Exception):
@@ -263,9 +263,8 @@ def _fd_trip(phase: str) -> None:
 
 
 def _deliver_fd_transcript(text: str) -> None:
-    """Emit the captured interjection; a bare stop phrase also ends the voice chat."""
-    # Stop-check must never break transcript delivery (stubbed voice_mode in tests,
-    # partial installs) — treat as not-a-stop.
+    """Emit the captured interjection; a bare stop phrase also ends the voice chat. The stop
+    check must never break delivery (stubbed voice_mode in tests, partial installs)."""
     try:
         from tools.voice_mode import is_voice_stop_phrase
         is_stop = is_voice_stop_phrase(text)
@@ -290,8 +289,7 @@ def _speak_text_with_barge(text: str) -> None:
     def _speak():
         try:
             speak_text(text, stop)
-        except TypeError:
-            # Older wrapper without the stop_event parameter.
+        except TypeError:  # older wrapper without the stop_event parameter
             speak_text(text)
         finally:
             done.set()
@@ -313,10 +311,13 @@ def _voice_cfg_number(value, default):
     return value if isinstance(value, (int, float)) and not isinstance(value, bool) else default
 
 
-def _voice_record_key() -> str:
-    """Current ``voice.record_key`` value, documented default on error."""
+def _voice_status_payload(**extra) -> dict:
+    """``{enabled, record_key, tts, **extra}``; record_key (documented default ``ctrl+b``) on every
+    voice.toggle branch so a tts toggle never resets a custom binding."""
     record_key = _voice_cfg_dict().get("record_key")
-    return str(record_key) if isinstance(record_key, str) and record_key else "ctrl+b"
+    record_key = str(record_key) if isinstance(record_key, str) and record_key else "ctrl+b"
+    return {"enabled": _voice_mode_enabled(), "record_key": record_key, "tts": _voice_tts_enabled(),
+            **extra}
 
 
 # ── Wake word ("Hey Hermes"): process-global detector (one mic). The first eligible transport
@@ -333,12 +334,6 @@ def _wake_owner_snapshot():
         return _wake_owner_transport, _wake_owner_surface
 
 
-def _set_wake_owner(transport, surface: str) -> None:
-    global _wake_owner_transport, _wake_owner_surface
-    with _wake_lock:
-        _wake_owner_transport, _wake_owner_surface = transport, surface
-
-
 def _release_wake_for_transport(transport: "Transport") -> bool:
     """Release the wake lease iff ``transport`` is the current gateway owner."""
     global _wake_owner_transport, _wake_owner_surface
@@ -387,14 +382,10 @@ def _wake_resume_if_owner(owner: "Transport", *, retry_seconds: float = 15.0,
         try:
             while time.monotonic() < deadline:
                 time.sleep(retry_interval)
-                try:
+                with contextlib.suppress(Exception):
                     if resume_listening(owner=owner):
                         logger.info("wake: detector resumed after retry")
-                        return
-                except Exception:
-                    continue
-                # False — detector gone or lease moved: stop, don't fight it.
-                return
+                    return  # False — detector gone or lease moved: stop, don't fight it.
             logger.warning("wake: could not resume detector after voice turn "
                            "(microphone still busy?) — toggle the wake word to re-arm")
         finally:
@@ -414,18 +405,15 @@ def _persist_wake_enabled(enabled: bool) -> bool:
         return False
 
 
-def _wake_prefers_client(params: dict, surface: str) -> bool:
-    """Desktop (gui) prefers client capture (Mac mic → wake.feed PCM); CLI/TUI stay local."""
-    return surface in ("gui", "desktop") or bool(params.get("client_capture"))
-
-
 def _frame_fields(frame: dict) -> dict:
     return {"sample_rate": frame.get("sample_rate", 16000), "frame_length": frame.get("frame_length", 1280)}
 
 
-def _wake_probe(cfg: dict, prefer_client: bool) -> tuple[str, dict]:
-    """``(capture_mode, requirements)``; capture stamped so the probe matches what would arm."""
+def _wake_probe(cfg: dict, params: dict, surface: str) -> tuple[str, dict]:
+    """``(capture_mode, requirements)``; capture stamped so the probe matches what would arm.
+    Desktop (gui) prefers client capture (Mac mic → wake.feed PCM); CLI/TUI stay local."""
     from tools.wake_word import check_wake_word_requirements, resolve_capture_mode
+    prefer_client = surface in ("gui", "desktop") or bool(params.get("client_capture"))
     capture_mode = resolve_capture_mode(cfg, prefer_client=prefer_client)
     return capture_mode, check_wake_word_requirements({**cfg, "capture": capture_mode})
 
@@ -473,8 +461,8 @@ def _(rid, params: dict) -> dict:
     """Arm the wake-word listener for the calling surface ("tui" | "gui"); ``{started: False,
     reason}`` when disabled, owned by another surface, or deps/mic aren't ready. ``persist: true``
     (explicit gesture) flips ``wake_word.enabled`` on before arming; auto-arm callers omit it."""
+    global _wake_owner_transport, _wake_owner_surface
     surface = str(params.get("surface") or "auto").strip().lower()
-    persist = bool(params.get("persist"))
     transport = _caller_transport()
     try:
         from tools.wake_word import (
@@ -483,16 +471,14 @@ def _(rid, params: dict) -> dict:
     except Exception as e:
         return _err(rid, 5026, f"wake module unavailable: {e}")
     cfg = load_wake_word_config()
-    capture_mode, reqs = _wake_probe(cfg, _wake_prefers_client(params, surface))
-    external_audio = capture_mode == "client"
+    capture_mode, reqs = _wake_probe(cfg, params, surface)
     # Requirements first: a gesture on an un-armable setup must refuse WITHOUT flipping
     # wake_word.enabled — else config says on while nothing can arm.
     if not reqs["available"]:
         logger.warning("wake.start(%s): not available — %s", surface, reqs.get("hint"))
-        return _ok(rid, {
-            "started": False, "reason": "unavailable", "hint": reqs.get("hint") or "",
-            "capture": capture_mode})
-    enabled_persisted = bool(persist and not cfg.get("enabled") and _persist_wake_enabled(True))
+        return _ok(rid, {"started": False, "reason": "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))
     if enabled_persisted:
         cfg = {**cfg, "enabled": True}
     if not wake_surface_enabled(surface, cfg):
@@ -503,9 +489,7 @@ def _(rid, params: dict) -> dict:
                     surface, reason, cfg.get("enabled"), cfg.get("surface"))
         return _ok(rid, {"started": False, "reason": 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:
@@ -514,18 +498,17 @@ def _(rid, params: dict) -> dict:
     try:
         on_detect = _wake_detect_handler(
             transport, sid, wake_phrase(cfg), bool(cfg.get("start_new_session", True)))
-        start_listening(on_detect, owner=transport, config=cfg, external_audio=external_audio)
+        start_listening(on_detect, owner=transport, config=cfg, external_audio=capture_mode == "client")
     except WakeWordInUse:
-        return _ok(rid, {"started": False, "reason": "owned",
-                         "owner_surface": existing_surface or None})
+        return _ok(rid, {"started": False, "reason": "owned", "owner_surface": existing_surface or None})
     except Exception as e:
         logger.warning("wake.start(%s): failed to start listener: %s", surface, e)
         return _err(rid, 5026, str(e))
-    _set_wake_owner(transport, surface)
+    with _wake_lock:
+        _wake_owner_transport, _wake_owner_surface = transport, surface
     frame = detector_frame_info()
-    logger.info(
-        "wake.start(%s): listening for %r (%s) capture=%s frame=%s",
-        surface, reqs["phrase"], reqs["provider"], capture_mode, frame.get("frame_length"))
+    logger.info("wake.start(%s): listening for %r (%s) capture=%s frame=%s",
+                surface, reqs["phrase"], reqs["provider"], capture_mode, frame.get("frame_length"))
     return _ok(rid, {
         "started": True, "phrase": reqs["phrase"], "provider": reqs["provider"],
         "owner_surface": surface, "enabled_persisted": enabled_persisted, "capture": capture_mode,
@@ -535,8 +518,7 @@ def _(rid, params: dict) -> dict:
 @method("wake.stop")
 def _(rid, params: dict) -> dict:
     """Stop this surface's listener; ``persist: true`` also writes ``wake_word.enabled: false``."""
-    transport = _caller_transport()
-    stopped = _release_wake_for_transport(transport)
+    stopped = _release_wake_for_transport(_caller_transport())
     disabled_persisted = False
     if bool(params.get("persist")):
         try:
@@ -546,18 +528,16 @@ def _(rid, params: dict) -> dict:
             currently_enabled = True
         if currently_enabled:
             disabled_persisted = _persist_wake_enabled(False)
-    return _ok(rid, {
-        "stopped": stopped, "reason": None if stopped else "not_owner",
-        "disabled_persisted": disabled_persisted})
+    return _ok(rid, {"stopped": stopped, "reason": None if stopped else "not_owner",
+                     "disabled_persisted": disabled_persisted})
 
 
 @method("wake.pause")
 def _(rid, params: dict) -> dict:
     """Release the mic (e.g. while the desktop's browser captures audio)."""
-    transport = _caller_transport()
     try:
         from tools.wake_word import pause_listening
-        paused = pause_listening(owner=transport)
+        paused = pause_listening(owner=_caller_transport())
         logger.info("wake.pause: detector paused=%s", paused)
     except Exception as e:
         logger.debug("wake.pause failed: %s", e)
@@ -580,11 +560,9 @@ def _(rid, params: dict) -> dict:
             audio_is_silent, detector_frame_info, get_input_device_status, is_listening,
             load_wake_word_config, owns_listener, silent_audio_hint)
         cfg = load_wake_word_config()
-        surface = str(params.get("surface") or "").strip().lower()
-        probe_capture, reqs = _wake_probe(cfg, _wake_prefers_client(params, surface))
-        transport = _caller_transport()
+        probe_capture, reqs = _wake_probe(cfg, params, str(params.get("surface") or "").strip().lower())
         owner, owner_surface = _wake_owner_snapshot()
-        owned_by_caller = owns_listener(transport)
+        owned_by_caller = owns_listener(_caller_transport())
         listening = owned_by_caller and is_listening()
         silent = listening and audio_is_silent()
         input_device = get_input_device_status(cfg)
@@ -621,7 +599,6 @@ def _(rid, params: dict) -> dict:
 def _(rid, params: dict) -> dict:
     """Push client-captured PCM (``pcm``/``pcm_b64``: base64 int16 mono LE, 16 kHz only) into the
     armed detector (``capture: "client"``) so mic-less remote backends can run openWakeWord."""
-    transport = _caller_transport()
     raw_b64 = params.get("pcm") or params.get("pcm_b64") or ""
     if not isinstance(raw_b64, str) or not raw_b64.strip():
         return _err(rid, 4001, "wake.feed requires base64 pcm")
@@ -632,15 +609,14 @@ def _(rid, params: dict) -> dict:
         return _err(rid, 4001, f"invalid base64 pcm: {e}")
     if not pcm:
         return _ok(rid, {"fed": False, "reason": "empty"})
-    # Soft size cap: 64000 bytes = 2s of 16 kHz int16 mono
-    if len(pcm) > 64000:
+    if len(pcm) > 64000:  # soft cap: 2s of 16 kHz int16 mono
         return _err(rid, 4001, "pcm frame too large")
     sr = params.get("sample_rate")
     if sr is not None and int(sr) not in (0, 16000):
         return _err(rid, 4001, "wake.feed only accepts 16 kHz PCM")
     try:
         from tools.wake_word import feed_audio
-        ok = feed_audio(owner=transport, pcm_int16=pcm)
+        ok = feed_audio(owner=_caller_transport(), pcm_int16=pcm)
     except Exception as e:
         logger.debug("wake.feed failed: %s", e)
         return _err(rid, 5026, str(e))
@@ -650,8 +626,7 @@ def _(rid, params: dict) -> dict:
 def _voice_toggle_status(rid, params: dict) -> dict:
     # Mirrors CLI _show_voice_status: STT/TTS availability tells the user WHY voice isn't
     # working; record_key lets the TUI bind and display the shortcut.
-    payload: dict = {"enabled": _voice_mode_enabled(), "record_key": _voice_record_key(),
-                     "tts": _voice_tts_enabled()}
+    payload = _voice_status_payload()
     try:
         from tools.voice_mode import check_voice_requirements
         reqs = check_voice_requirements()
@@ -667,16 +642,13 @@ def _voice_toggle_status(rid, params: dict) -> dict:
 
 def _voice_toggle_mode(rid, params: dict) -> dict:
     enabled = params.get("action") == "on"
-    # Runtime-only flag — never persisted, so the next launch starts with voice OFF.
     os.environ["HERMES_VOICE"] = "1" if enabled else "0"
     stop_hint = ""
     if enabled:
         # Spoken-stop hint for the client; sourced from voice.stop_phrases, empty when disabled.
-        try:
+        with contextlib.suppress(Exception):
             from tools.voice_mode import voice_stop_hint
             stop_hint = voice_stop_hint()
-        except Exception:
-            stop_hint = ""
         # Speech output already on → warm the engine now, not on the first reply.
         if _voice_tts_enabled():
             _tts_lease_async("tui:voice-tts", True)
@@ -693,8 +665,7 @@ def _voice_toggle_mode(rid, params: dict) -> dict:
         os.environ["HERMES_VOICE_TTS"] = "0"
         _tts_stream_stop(user_barge=False)
         _tts_lease_async("tui:voice-tts", False)
-    return _ok(rid, {"enabled": enabled, "record_key": _voice_record_key(),
-                     "tts": _voice_tts_enabled(), "stop_hint": stop_hint})
+    return _ok(rid, _voice_status_payload(stop_hint=stop_hint))
 
 
 def _voice_toggle_tts(rid, params: dict) -> dict:
@@ -706,8 +677,7 @@ def _voice_toggle_tts(rid, params: dict) -> dict:
         _tts_stream_stop(user_barge=False)
     # on → pre-load the engine so the first reply starts hot; off → release the lease.
     _tts_lease_async("tui:voice-tts", new_value)
-    # record_key on every branch so a tts toggle never resets a custom binding.
-    return _ok(rid, {"enabled": True, "record_key": _voice_record_key(), "tts": new_value})
+    return _ok(rid, _voice_status_payload())
 
 
 _VOICE_TOGGLE_ACTIONS = {
@@ -726,7 +696,8 @@ def _(rid, params: dict) -> dict:
     return handler(rid, params)
 
 
-# voice.record callbacks (module-level: they touch only process-global state).
+# voice.record callbacks (module-level: they touch only process-global state). Each terminal
+# capture event resumes the wake detector so wake-triggered and manual captures coexist.
 def _vr_on_transcript(t):
     _voice_emit("voice.transcript", {"text": t})
     _resume_voice_wake()
@@ -751,12 +722,50 @@ def _vr_on_status(state):
         _resume_voice_wake()
 
 
+def _voice_record_start(transport) -> dict:
+    """Start the VAD-bounded capture; returns the result payload. If the wake detector handed over
+    the mic, a terminal capture event (or a failed start) resumes it."""
+    global _voice_wake_owner
+    from hermes_cli.voice import start_continuous
+    # Busy probe holds the no-speech counter during long agent turns.
+    # Safe to re-register every start; older wrappers lack the setter.
+    with contextlib.suppress(Exception):
+        from hermes_cli.voice import set_voice_busy_probe
+        set_voice_busy_probe(_any_session_running)
+    # Shape-safe: malformed voice YAML falls back to documented defaults.
+    # max_recording_seconds: explicit numeric <= 0 disables the cap (0.0).
+    voice_cfg = _voice_cfg_dict()
+    max_rec = _voice_cfg_number(voice_cfg.get("max_recording_seconds"), 120.0)
+    try:
+        from tools.wake_word import pause_listening
+        wake_paused = pause_listening(owner=transport)
+    except Exception:
+        wake_paused = False
+    if wake_paused:
+        with _voice_sid_lock:
+            _voice_wake_owner = transport
+    try:
+        started = start_continuous(
+            on_transcript=_vr_on_transcript, on_status=_vr_on_status, on_silent_limit=_vr_on_silent,
+            silence_threshold=_voice_cfg_number(voice_cfg.get("silence_threshold"), 200),
+            silence_duration=_voice_cfg_number(voice_cfg.get("silence_duration"), 3.0),
+            auto_restart=False, max_recording_seconds=max_rec if max_rec > 0 else 0.0,
+            on_stop_phrase=_vr_on_stop_phrase)
+    except Exception:
+        if wake_paused:
+            _resume_voice_wake()
+        raise
+    if started is False:
+        _resume_voice_wake()
+        return {"status": "busy"}
+    return {"status": "recording"}
+
+
 @method("voice.record")
 def _(rid, params: dict) -> dict:
     """VAD-bounded push-to-talk. ``start`` emits ``voice.transcript`` when silence stops the
     capture; ``stop`` forces transcription. Three silent captures emit ``no_speech_limit``."""
     action = params.get("action", "start")
-    wake_paused = False
     if action not in {"start", "stop"}:
         return _err(rid, 4019, f"unknown voice action: {action}")
     transport = _caller_transport()
@@ -764,7 +773,7 @@ def _(rid, params: dict) -> dict:
     if wake_owner is not None and wake_owner is not transport:
         return _ok(rid, {"status": "busy", "reason": "wake_owned"})
     try:
-        global _voice_event_sid, _voice_wake_owner
+        global _voice_event_sid
         if action == "start" and not _voice_mode_enabled():
             return _err(rid, 4015, "voice mode is off — enable with /voice on")
         with _voice_sid_lock:
@@ -774,39 +783,9 @@ def _(rid, params: dict) -> dict:
             stop_continuous(force_transcribe=True)
             _resume_voice_wake()
             return _ok(rid, {"status": "stopped"})
-        from hermes_cli.voice import start_continuous
-        # Busy probe holds the no-speech counter during long agent turns.
-        # Safe to re-register every start; older wrappers lack the setter.
-        with contextlib.suppress(Exception):
-            from hermes_cli.voice import set_voice_busy_probe
-            set_voice_busy_probe(_any_session_running)
-        # Shape-safe: malformed voice YAML falls back to documented defaults.
-        # max_recording_seconds: explicit numeric <= 0 disables the cap (0.0).
-        voice_cfg = _voice_cfg_dict()
-        max_rec = _voice_cfg_number(voice_cfg.get("max_recording_seconds"), 120.0)
-        # Hand the mic to STT if the wake detector holds it; resume on a terminal
-        # capture event so wake-triggered and manual captures coexist.
-        try:
-            from tools.wake_word import pause_listening
-            wake_paused = pause_listening(owner=transport)
-        except Exception:
-            wake_paused = False
-        if wake_paused:
-            with _voice_sid_lock:
-                _voice_wake_owner = transport
-        started = start_continuous(
-            on_transcript=_vr_on_transcript, on_status=_vr_on_status, on_silent_limit=_vr_on_silent,
-            silence_threshold=_voice_cfg_number(voice_cfg.get("silence_threshold"), 200),
-            silence_duration=_voice_cfg_number(voice_cfg.get("silence_duration"), 3.0),
-            auto_restart=False, max_recording_seconds=max_rec if max_rec > 0 else 0.0,
-            on_stop_phrase=_vr_on_stop_phrase,
-        )
-        if started is False:
-            _resume_voice_wake()
-            return _ok(rid, {"status": "busy"})
-        return _ok(rid, {"status": "recording"})
+        return _ok(rid, _voice_record_start(transport))
     except Exception as e:
-        if wake_paused or action == "stop":
+        if action == "stop":
             _resume_voice_wake()
         if isinstance(e, ImportError):
             return _err(rid, 5025, "voice module not available — install audio dependencies")
@@ -819,8 +798,7 @@ def _(rid, params: dict) -> dict:
     if not text:
         return _err(rid, 4020, "text required")
     try:
-        # Import check up front so a missing voice module returns 5026, not a silent thread death.
-        import hermes_cli.voice  # noqa: F401
+        import hermes_cli.voice  # noqa: F401  (a missing module must answer 5026, not die in a thread)
     except ImportError:
         return _err(rid, 5026, "voice module not available")
     except Exception as e: