From 7532f2045f1836b25c7c0965b1235c10fadc55ab Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:12:13 -0700
Subject: [PATCH] =?UTF-8?q?refactor(tui=5Fgateway):=20methods=5Ftools/slas?=
 =?UTF-8?q?h/complete=20=E2=80=94=20session+error=20guards,=20shared=20plu?=
 =?UTF-8?q?gin=20runner,=20learning.*=20table,=20layout=20compaction?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tui_gateway/methods_complete.py         | 200 ++++-----
 tui_gateway/methods_complete_helpers.py |  34 +-
 tui_gateway/methods_slash.py            | 113 ++----
 tui_gateway/methods_tools.py            | 517 ++++++++----------------
 4 files changed, 296 insertions(+), 568 deletions(-)

diff --git a/tui_gateway/methods_complete.py b/tui_gateway/methods_complete.py
index 60f47469db..7f2869070c 100644
--- a/tui_gateway/methods_complete.py
+++ b/tui_gateway/methods_complete.py
@@ -18,14 +18,12 @@ _AT_DIRECTIVE_HINTS = [
     ("@file:", "attach file"),
     ("@folder:", "attach folder"),
     ("@url:", "fetch url"),
-    ("@git:", "git log"),
-]
+    ("@git:", "git log")]
 _SLASH_EXTRAS = [
     ("/density", "Toggle compact display mode"),
     ("/details", "Control agent detail visibility"),
     ("/logs", "Show recent gateway log lines"),
-    ("/mouse", "Set mouse tracking preset [on|off|toggle|wheel|buttons|all]"),
-]
+    ("/mouse", "Set mouse tracking preset [on|off|toggle|wheel|buttons|all]")]
 
 
 def _item(text: str, meta: str, display: str | None = None) -> dict:
@@ -38,17 +36,13 @@ def _(rid, params: dict) -> dict:
     text = params.get("text", "")
     if not text:
         return _err(rid, 4004, "empty paste")
-
     _paste_counter += 1
     line_count = text.count("\n") + 1
     paste_dir = _hermes_home / "pastes"
     paste_dir.mkdir(parents=True, exist_ok=True)
-
     from datetime import datetime
-
     paste_file = paste_dir / f"paste_{_paste_counter}_{datetime.now().strftime('%H%M%S')}.txt"
     paste_file.write_text(text, encoding="utf-8")
-
     placeholder = f"[Pasted text #{_paste_counter}: {line_count} lines \u2192 {paste_file}]"
     return _ok(rid, {"placeholder": placeholder, "path": str(paste_file), "lines": line_count})
 
@@ -60,7 +54,6 @@ def _profile_mention_items(prefix: str) -> list[dict]:
     out: list[dict] = []
     try:
         from hermes_cli.profiles import list_profiles
-
         seen: set[str] = set()
         for p in list_profiles():
             name = (p.name or "").strip()
@@ -82,12 +75,10 @@ def _plugin_reference_items(pfx: str, qval: str) -> list[dict] | None:
     no provider owns ``pfx`` or it fails."""
     try:
         from agent.context_references import get_context_reference_providers
-
         prov = get_context_reference_providers().get(pfx)
         if prov is None:
             return None
         import asyncio
-
         coro = prov.autocomplete(qval, limit=20)
         try:
             loop = asyncio.get_running_loop()
@@ -95,7 +86,6 @@ def _plugin_reference_items(pfx: str, qval: str) -> list[dict] | None:
             loop = None
         if loop and loop.is_running():
             import concurrent.futures
-
             with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                 ac = pool.submit(asyncio.run, coro).result()
         else:
@@ -128,7 +118,6 @@ def _fuzzy_basename_items(root: str, path_part: str, prefix_tag: str) -> list[di
                 _consider(entry, entry, os.path.isdir(os.path.join(root, entry)))
     except OSError:
         pass
-
     for rel in _list_repo_files(root):
         _consider(rel, os.path.basename(rel), False)
         # Rank each ancestor dir too — a folder with no name-matching file inside is otherwise invisible.
@@ -145,10 +134,8 @@ def _fuzzy_basename_items(root: str, path_part: str, prefix_tag: str) -> list[di
         _item(
             f"@{'folder' if is_dir else tag}:{rel}{'/' if is_dir else ''}",
             "dir" if is_dir else os.path.dirname(rel),
-            basename + ("/" if is_dir else ""),
-        )
-        for _, rel, basename, is_dir in ranked[:30]
-    ]
+            basename + ("/" if is_dir else ""))
+        for _, rel, basename, is_dir in ranked[:30]]
 
 
 @method("complete.path")
@@ -156,19 +143,16 @@ def _(rid, params: dict) -> dict:
     word = params.get("word", "")
     if not word:
         return _ok(rid, {"items": []})
-
     items: list[dict] = []
     try:
         root = _completion_cwd(params)
         is_context = word.startswith("@")
         query = word[1:] if is_context else word
-
         if is_context and not query:
             items = [_item(t, m) for t, m in _AT_DIRECTIVE_HINTS]
             items.extend(_profile_mention_items(""))  # `@` alone reveals agent profiles too
             try:
                 from agent.context_references import get_context_reference_providers
-
                 for _pfx, _prov in sorted(get_context_reference_providers().items()):
                     items.append(_item(f"@{_pfx}:", _prov.description or f"plugin: {_pfx}"))
             except Exception:
@@ -196,13 +180,11 @@ def _(rid, params: dict) -> dict:
         if is_context and path_part.startswith("/") and not path_part.startswith("//"):
             if not _abs_completion_prefix_exists(path_part):
                 path_part = path_part.lstrip("/")
-
         if is_context and path_part and len(path_part.strip()) >= 2 and "/" not in path_part and prefix_tag != "folder":
             items = _fuzzy_basename_items(root, path_part, prefix_tag)
             if not prefix_tag:  # bare `@name` may be an agent mention: profiles rank ABOVE file hits
                 items = _profile_mention_items(path_part) + items
             return _ok(rid, {"items": items})
-
         expanded = _normalize_completion_path(path_part) if path_part else "."
         if expanded == "." or not expanded:
             search_dir, match = ".", ""
@@ -211,11 +193,9 @@ def _(rid, params: dict) -> dict:
         else:
             search_dir = os.path.dirname(expanded) or "."
             match = os.path.basename(expanded)
-
         search_dir = search_dir if os.path.isabs(search_dir) else os.path.join(root, search_dir)
         if not os.path.isdir(search_dir):
             return _ok(rid, {"items": []})
-
         want_dir = prefix_tag == "folder"
         match_lower = match.lower()
         for entry in sorted(os.listdir(search_dir)):
@@ -232,7 +212,6 @@ def _(rid, params: dict) -> dict:
                 continue
             rel = os.path.relpath(full, root).replace(os.sep, "/")
             suffix = "/" if is_dir else ""
-
             if is_context and prefix_tag:
                 text = f"@{prefix_tag}:{rel}{suffix}"
             elif is_context:
@@ -243,7 +222,6 @@ def _(rid, params: dict) -> dict:
                 text = "./" + rel + suffix
             else:
                 text = rel + suffix
-
             items.append(_item(text, "dir" if is_dir else "", entry + suffix))
             if len(items) >= 30:
                 break
@@ -256,7 +234,6 @@ def _(rid, params: dict) -> dict:
             items = _profile_mention_items(path_part) + items
     except Exception:
         pass
-
     return _ok(rid, {"items": items})
 
 
@@ -265,15 +242,12 @@ def _(rid, params: dict) -> dict:
     text = params.get("text", "")
     if not text.startswith("/"):
         return _ok(rid, {"items": []})
-
     try:
         from hermes_cli.commands import SlashCommandCompleter
         from prompt_toolkit.document import Document
         from prompt_toolkit.formatted_text import to_plain_text
-
         from agent.skill_commands import get_skill_commands
         from agent.skill_bundles import get_skill_bundles
-
         completer = SlashCommandCompleter(
             skill_commands_provider=lambda: get_skill_commands(), skill_bundles_provider=lambda: get_skill_bundles()
         )
@@ -291,9 +265,7 @@ def _(rid, params: dict) -> dict:
                     "meta": to_plain_text(c.display_meta) if c.display_meta else "",
                     "kind": "skill" if c.text.strip().lstrip("/").lower() in skill_names else "command",
                 }
-                for c in completer.get_completions(doc, None)
-            ]
-
+                for c in completer.get_completions(doc, None)]
         items = to_items(Document(text, len(text)))
 
         # Rank + bound while a `/token` is under the cursor (the one stage skills are
@@ -304,120 +276,114 @@ def _(rid, params: dict) -> dict:
             # catalog entries whose name SUBSTRING or DESCRIPTION words match (name outranks description).
             if " " not in text and len(text) > 1:
                 from tui_gateway.slash_fuzzy import fuzzy_rank_slash_items, normalize_slash_search_query
-
                 items, score_of = fuzzy_rank_slash_items(
-                    items, to_items(Document("/", 1)), normalize_slash_search_query(text)
-                )
-
+                    items, to_items(Document("/", 1)), normalize_slash_search_query(text))
             usage, origin_of = _skill_usage_lookup()
             items = _rank_slash_completions(items, usage, origin_of, browsing=text == "/", score_of=score_of)
         else:
             items = items[:_SLASH_COMPLETION_LIMIT]
-
         text_lower = text.lower()
         for extra_text, extra_meta in _SLASH_EXTRAS:
             if extra_text.startswith(text_lower) and not any(item["text"] == extra_text for item in items):
                 items.append({"text": extra_text, "display": extra_text, "meta": extra_meta, "kind": "command"})
-
         details_items = _details_completions(text)
         if details_items is not None:
             return _ok(rid, {"items": details_items, "replace_from": text.rfind(" ") + 1 if " " in text else len(text)})
-
         return _ok(rid, {"items": items, "replace_from": text.rfind(" ") + 1 if " " in text else 1})
     except Exception as e:
         return _err(rid, 5020, str(e))
 
 
+def _catch(fail_code: int):
+    """Handler body exceptions → ``_err(rid, fail_code, str(e))``."""
+
+    def deco(body):
+        def handler(rid, params: dict) -> dict:
+            try:
+                return body(rid, params)
+            except Exception as e:
+                return _err(rid, fail_code, str(e))
+
+        handler.__doc__ = body.__doc__
+        return handler
+
+    return deco
+
+
+def _session_agent(params: dict):
+    session = _sessions.get(params.get("session_id", ""))
+    return session.get("agent") if session else None
+
+
 @method("model.options")
 @_profile_scoped
+@_catch(5033)
 def _(rid, params: dict) -> dict:
-    try:
-        from hermes_cli.inventory import build_model_options_payload
-
-        session = _sessions.get(params.get("session_id", ""))
-        agent = session.get("agent") if session else None
-        # A spawned agent owns the live provider/model/base_url; empty attributes must
-        # NOT clobber disk config (with_overrides is truthy-only).
-        ctx = _model_picker_context(agent)
-        payload = build_model_options_payload(
-            ctx,
-            explicit_only=bool(params.get("explicit_only")),
-            include_unconfigured=bool(params.get("include_unconfigured")),
-            refresh=bool(params.get("refresh")),
-        )
-        return _ok(rid, payload)
-    except Exception as e:
-        return _err(rid, 5033, str(e))
+    from hermes_cli.inventory import build_model_options_payload
+    # A spawned agent owns the live provider/model/base_url; empty attributes must
+    # NOT clobber disk config (with_overrides is truthy-only).
+    ctx = _model_picker_context(_session_agent(params))
+    payload = build_model_options_payload(
+        ctx,
+        explicit_only=bool(params.get("explicit_only")),
+        include_unconfigured=bool(params.get("include_unconfigured")),
+        refresh=bool(params.get("refresh")))
+    return _ok(rid, payload)
 
 
 @method("model.save_key")
+@_catch(5034)
 def _(rid, params: dict) -> dict:
     """Save an API key for ``slug``; return its refreshed provider row (model.options shape + ``authenticated``)."""
-    try:
-        from hermes_cli.auth import PROVIDER_REGISTRY
-        from hermes_cli.config import is_managed
-        from hermes_cli.inventory import build_models_payload
-
-        slug = (params.get("slug") or "").strip()
-        api_key = (params.get("api_key") or "").strip()
-        if not slug or not api_key:
-            return _err(rid, 4001, "slug and api_key are required")
-        if is_managed():
-            return _err(rid, 4006, "managed install — credentials are read-only")
-        pconfig = PROVIDER_REGISTRY.get(slug)
-        if not pconfig:
-            return _err(rid, 4002, f"unknown provider: {slug}")
-        if pconfig.auth_type != "api_key":
-            return _err(rid, 4003, f"{pconfig.name} uses {pconfig.auth_type} auth — run `hermes model` to configure")
-        if not pconfig.api_key_env_vars:
-            return _err(rid, 4004, f"no env var defined for {pconfig.name}")
-
-        # Unified lifecycle rotates stale config.yaml mirrors of the old key too.
-        env_var = pconfig.api_key_env_vars[0]
-        from hermes_cli.credential_lifecycle import save_provider_env_credential
-
-        save_provider_env_credential(env_var, api_key)
-        os.environ[env_var] = api_key  # so the refreshed inventory sees it
-
-        # Shared inventory builder (lock-step with model.options / dashboard); picker_hints carries `authenticated`.
-        session = _sessions.get(params.get("session_id", ""))
-        agent = session.get("agent") if session else None
-        payload = build_models_payload(_model_picker_context(agent), picker_hints=True, max_models=50)
-        provider_data = next((p for p in payload["providers"] if p["slug"] == slug), None)
-        if provider_data is None:  # key saved but provider didn't appear — still success
-            provider_data = {"slug": slug, "name": pconfig.name, "is_current": False, "models": [], "total_models": 0}
-        provider_data["authenticated"] = True  # synthetic fallback bypasses picker_hints
-        return _ok(rid, {"provider": provider_data})
-    except Exception as e:
-        return _err(rid, 5034, str(e))
+    from hermes_cli.auth import PROVIDER_REGISTRY
+    from hermes_cli.config import is_managed
+    from hermes_cli.inventory import build_models_payload
+    slug = (params.get("slug") or "").strip()
+    api_key = (params.get("api_key") or "").strip()
+    if not slug or not api_key:
+        return _err(rid, 4001, "slug and api_key are required")
+    if is_managed():
+        return _err(rid, 4006, "managed install — credentials are read-only")
+    pconfig = PROVIDER_REGISTRY.get(slug)
+    if not pconfig:
+        return _err(rid, 4002, f"unknown provider: {slug}")
+    if pconfig.auth_type != "api_key":
+        return _err(rid, 4003, f"{pconfig.name} uses {pconfig.auth_type} auth — run `hermes model` to configure")
+    if not pconfig.api_key_env_vars:
+        return _err(rid, 4004, f"no env var defined for {pconfig.name}")
+    # Unified lifecycle rotates stale config.yaml mirrors of the old key too.
+    env_var = pconfig.api_key_env_vars[0]
+    from hermes_cli.credential_lifecycle import save_provider_env_credential
+    save_provider_env_credential(env_var, api_key)
+    os.environ[env_var] = api_key  # so the refreshed inventory sees it
+    # Shared inventory builder (lock-step with model.options / dashboard); picker_hints carries `authenticated`.
+    payload = build_models_payload(_model_picker_context(_session_agent(params)), picker_hints=True, max_models=50)
+    provider_data = next((p for p in payload["providers"] if p["slug"] == slug), None)
+    if provider_data is None:  # key saved but provider didn't appear — still success
+        provider_data = {"slug": slug, "name": pconfig.name, "is_current": False, "models": [], "total_models": 0}
+    provider_data["authenticated"] = True  # synthetic fallback bypasses picker_hints
+    return _ok(rid, {"provider": provider_data})
 
 
 @method("model.disconnect")
+@_catch(5035)
 def _(rid, params: dict) -> dict:
     """Remove all credentials (env keys AND OAuth/pool state) for provider ``slug``."""
-    try:
-        from hermes_cli.auth import PROVIDER_REGISTRY, clear_provider_auth
-        from hermes_cli.credential_lifecycle import remove_provider_env_credential
-
-        slug = (params.get("slug") or "").strip()
-        if not slug:
-            return _err(rid, 4001, "slug is required")
-        pconfig = PROVIDER_REGISTRY.get(slug)
-        cleared_env = False
-        # Remove env vars plus every mirror (env-seeded pool entries, model cache rows,
-        # value-matched config.yaml copies) or the provider resurrects in the picker after restart.
-        if pconfig and pconfig.api_key_env_vars:
-            for ev in pconfig.api_key_env_vars:
-                if remove_provider_env_credential(ev).get("found"):
-                    cleared_env = True
-
-        # Full disconnect: removing OAuth grants is intended here, unlike key-only deletes.
-        cleared_auth = clear_provider_auth(slug)
-        if not cleared_env and not cleared_auth:
-            return _err(rid, 4005, f"no credentials found for {slug}")
-        return _ok(rid, {"slug": slug, "name": pconfig.name if pconfig else slug, "disconnected": True})
-    except Exception as e:
-        return _err(rid, 5035, str(e))
+    from hermes_cli.auth import PROVIDER_REGISTRY, clear_provider_auth
+    from hermes_cli.credential_lifecycle import remove_provider_env_credential
+    slug = (params.get("slug") or "").strip()
+    if not slug:
+        return _err(rid, 4001, "slug is required")
+    pconfig = PROVIDER_REGISTRY.get(slug)
+    # Remove EVERY env var plus its mirrors (env-seeded pool entries, model cache rows,
+    # value-matched config.yaml copies) or the provider resurrects in the picker after restart.
+    env_vars = (pconfig.api_key_env_vars if pconfig else None) or ()
+    cleared_env = any([remove_provider_env_credential(ev).get("found") for ev in env_vars])
+    # Full disconnect: removing OAuth grants is intended here, unlike key-only deletes.
+    cleared_auth = clear_provider_auth(slug)
+    if not cleared_env and not cleared_auth:
+        return _err(rid, 4005, f"no credentials found for {slug}")
+    return _ok(rid, {"slug": slug, "name": pconfig.name if pconfig else slug, "disconnected": True})
 
 
 def register(server) -> None:
diff --git a/tui_gateway/methods_complete_helpers.py b/tui_gateway/methods_complete_helpers.py
index 1e5df8c83c..b6631ab069 100644
--- a/tui_gateway/methods_complete_helpers.py
+++ b/tui_gateway/methods_complete_helpers.py
@@ -19,8 +19,7 @@ _FUZZY_CACHE_TTL_S = 5.0
 _FUZZY_CACHE_MAX_FILES = 20000
 _FUZZY_FALLBACK_EXCLUDES = frozenset(
     {".git", ".hg", ".svn", ".next", ".cache", ".venv", "venv", "node_modules", "__pycache__",
-     "dist", "build", "target", ".mypy_cache", ".pytest_cache", ".ruff_cache"}
-)
+     "dist", "build", "target", ".mypy_cache", ".pytest_cache", ".ruff_cache"})
 _fuzzy_cache_lock = threading.Lock()
 _fuzzy_cache: dict[str, tuple[float, list[str]]] = {}
 
@@ -35,10 +34,8 @@ def _list_repo_files(root: str) -> list[str]:
         cached = _fuzzy_cache.get(root)
         if cached and now - cached[0] < _FUZZY_CACHE_TTL_S:
             return cached[1]
-
     files: list[str] = []
     from hermes_cli._subprocess_compat import windows_hide_flags
-
     run_kw = dict(capture_output=True, timeout=2.0, check=False, stdin=subprocess.DEVNULL, creationflags=windows_hide_flags())
     try:
         top_result = subprocess.run(["git", "-C", root, "rev-parse", "--show-toplevel"], **run_kw)
@@ -59,7 +56,6 @@ def _list_repo_files(root: str) -> list[str]:
                         break
     except (OSError, subprocess.TimeoutExpired):
         pass
-
     if not files:
         # Fallback walk skips vendor/build dirs + dot-dirs; dotfiles survive (the ranker
         # decides based on whether the query starts with `.`).
@@ -76,10 +72,8 @@ def _list_repo_files(root: str) -> list[str]:
                     break
         except OSError:
             pass
-
     with _fuzzy_cache_lock:
         _fuzzy_cache[root] = (now, files)
-
     return files
 
 
@@ -89,13 +83,10 @@ def _fuzzy_basename_rank(name: str, query: str) -> tuple[int, int] | None:
     · 3 substring · 4 subsequence (query chars appear in order)."""
     if not query:
         return (3, len(name))
-
     nl = name.lower()
     ql = query.lower()
-
     if nl == ql:
         return (0, len(name))
-
     if nl.startswith(ql):
         return (1, len(name))
 
@@ -115,17 +106,14 @@ def _fuzzy_basename_rank(name: str, query: str) -> tuple[int, int] | None:
     for p in parts:
         if p.lower().startswith(ql):
             return (2, len(name))
-
     if ql in nl:
         return (3, len(name))
-
     i = 0
     for ch in nl:
         if ch == ql[i]:
             i += 1
             if i == len(ql):
                 return (4, len(name))
-
     return None
 
 
@@ -136,13 +124,10 @@ def _abs_completion_prefix_exists(path_part: str) -> bool:
     expanded = _normalize_completion_path(path_part)
     parent = os.path.dirname(expanded.rstrip("/")) or "/"
     tail = os.path.basename(expanded.rstrip("/"))
-
     if not os.path.isdir(parent):
         return False
-
     if not tail or expanded.endswith("/"):
         return os.path.isdir(expanded) or expanded == "/"
-
     try:
         tail_lower = tail.lower()
         return any(e.lower().startswith(tail_lower) for e in os.listdir(parent))
@@ -171,11 +156,9 @@ def _details_root_meta(candidate: str) -> str:
 def _details_completions(text: str) -> list[dict] | None:
     if not text.lower().startswith("/details"):
         return None
-
     stripped = text.strip()
     if stripped and not "/details".startswith(stripped.lower().split()[0]):
         return None
-
     body = text[len("/details") :]
     if body.startswith(" "):
         body = body[1:]
@@ -183,41 +166,33 @@ def _details_completions(text: str) -> list[dict] | None:
     has_trailing_space = text.endswith(" ")
     sections, modes = _DETAILS_SECTIONS, _DETAILS_MODES
     root_candidates = (*modes, "cycle", *sections)
-
     if not body or (len(parts) == 0 and has_trailing_space):
         return [_details_root_completion_item(c, _details_root_meta(c), not has_trailing_space) for c in root_candidates]
-
     if len(parts) == 1 and not has_trailing_space:
         prefix = parts[0].lower()
         return [
             _details_completion_item(c, _details_root_meta(c))
             for c in root_candidates
-            if c.startswith(prefix) and c != prefix
-        ]
-
+            if c.startswith(prefix) and c != prefix]
     section = parts[0].lower() if parts else ""
     if section not in sections:
         return []
 
     def section_meta(candidate: str) -> str:
         return f"clear {section} override" if candidate == "reset" else f"set {section}"
-
     if len(parts) == 1 and has_trailing_space:
         return [_details_completion_item(c, section_meta(c)) for c in (*modes, "reset")]
-
     if len(parts) == 2 and not has_trailing_space:
         prefix = parts[1].lower()
         return [
             _details_completion_item(c, section_meta(c)) for c in (*modes, "reset") if c.startswith(prefix) and c != prefix
         ]
-
     return []
 
 
 def _model_picker_context(agent):
     """Layer live session state onto config without losing custom identity."""
     from hermes_cli.inventory import load_picker_context
-
     ctx = load_picker_context()
     provider = getattr(agent, "provider", "") if agent else ""
     base_url = getattr(agent, "base_url", "") if agent else ""
@@ -225,16 +200,13 @@ def _model_picker_context(agent):
     if str(provider or "").strip().lower() == "custom":
         try:
             from hermes_cli.runtime_provider import canonical_custom_identity
-
             provider = (
                 canonical_custom_identity(
                     base_url=base_url or None, config_provider=ctx.current_provider, model=model or None
                 )
-                or provider
-            )
+                or provider)
         except Exception:
             logger.debug("custom provider identity recovery failed (model picker)", exc_info=True)
-
     return ctx.with_overrides(
         current_provider=provider, current_model=model or _resolve_model(), current_base_url=base_url
     )
diff --git a/tui_gateway/methods_slash.py b/tui_gateway/methods_slash.py
index b8155536e1..270da69700 100644
--- a/tui_gateway/methods_slash.py
+++ b/tui_gateway/methods_slash.py
@@ -43,15 +43,12 @@ def _format_live_review_output(session: Optional[dict], arg: str) -> str:
         return "Nothing to review yet — send a message first."
     if session.get("running"):
         return "session busy — wait for the current turn to finish, then /review"
-
     with session.get("history_lock") or contextlib.nullcontext():
         snapshot = list(session.get("history", []))
     if not snapshot:
         snapshot = list(getattr(agent, "_session_messages", None) or [])
-
     try:
         from agent.review_engine import format_dispatch_note, start_review
-
         result = start_review(agent, snapshot, arg or "")
     except ValueError as exc:
         return str(exc)
@@ -73,27 +70,23 @@ def _format_live_usage_output(session: dict) -> str:
 
     def n(key: str) -> str:
         return f"{int(usage.get(key) or 0):,}"
-
     lines = [
         "Session Token Usage",
         "────────────────────────────────────────",
         f"Model: {usage.get('model') or _metadata_mirror(session).get('model') or getattr(agent, 'model', '') or '(unknown)'}",
         f"Input tokens:                 {n('input')}",
-        f"Output tokens:                {n('output')}",
-    ]
+        f"Output tokens:                {n('output')}"]
     if int(usage.get("reasoning") or 0):
         lines.append(f"Reasoning tokens:             {n('reasoning')}")
     lines += [
         f"Prompt tokens:                {n('prompt')}",
         f"Completion tokens:            {n('completion')}",
         f"Total tokens:                 {n('total')}",
-        f"API calls:                    {n('calls')}",
-    ]
+        f"API calls:                    {n('calls')}"]
     if usage.get("context_max"):
         lines.append(
             f"Current context:              {n('context_used')} / {n('context_max')} "
-            f"({int(usage.get('context_percent') or 0)}%)"
-        )
+            f"({int(usage.get('context_percent') or 0)}%)")
     lines += [f"Messages:                     {message_count:,}", f"Compressions:                 {n('compressions')}"]
     return "\n".join(lines)
 
@@ -106,8 +99,7 @@ def _live_session_messages(session: dict) -> Optional[list]:
         if db is not None and session.get("session_key"):
             try:
                 return db.get_messages_as_conversation(
-                    session["session_key"], include_ancestors=True, include_row_ids=True
-                )
+                    session["session_key"], include_ancestors=True, include_row_ids=True)
             except Exception:
                 pass
     return None
@@ -140,8 +132,7 @@ def _format_live_prompt_output(session: dict) -> str:
         mirror.get("system_prompt")
         or getattr(agent, "ephemeral_system_prompt", None)
         or getattr(agent, "_cached_system_prompt", None)
-        or ""
-    )
+        or "")
     if not prompt:
         return "Current system prompt is not built yet; send a message first."
     return f"Current system prompt:\n{prompt}"
@@ -164,8 +155,7 @@ def _format_live_context_output(session: dict) -> str:
         roles[role] = roles.get(role, 0) + 1
     lines.append(
         f"  user: {roles.get('user', 0)}, assistant: {roles.get('assistant', 0)}, "
-        f"tool: {roles.get('tool', 0)}, system: {roles.get('system', 0)}"
-    )
+        f"tool: {roles.get('tool', 0)}, system: {roles.get('system', 0)}")
     model = mirror.get("model") or usage.get("model") or ""
     if model:
         lines.append(f"Model: {model}")
@@ -197,7 +187,6 @@ def _format_live_tools_output(session: dict) -> str:
 def _format_live_help_output() -> str:
     try:
         from hermes_cli.commands import COMMANDS_BY_CATEGORY
-
         lines = ["Available commands:", ""]
         for category, commands in COMMANDS_BY_CATEGORY.items():
             lines.append(f"{category}:")
@@ -238,8 +227,7 @@ _LIVE_SLASH_OUTPUT = {
     "clear": (None, lambda sid, s, a: "Screen clear is terminal-only; desktop/TUI chat left unchanged."),
     "models": (None, lambda sid, s, a: "Use /model to view or switch the current model; desktop users can also open the model picker."),
     "rename": (None, lambda sid, s, a: "Use /title <name> to rename this session."),
-    "effort": (None, lambda sid, s, a: "Use /reasoning <effort> to change reasoning effort."),
-}
+    "effort": (None, lambda sid, s, a: "Use /reasoning <effort> to change reasoning effort.")}
 
 
 def _live_slash_command_output(sid: str, session: Optional[dict], name: str, arg: str) -> Optional[str]:
@@ -270,21 +258,21 @@ def _live_slash_command_output(sid: str, session: Optional[dict], name: str, arg
 _MUTATES_WHILE_RUNNING = frozenset({"model", "personality", "prompt", "compress"})
 
 
-def _compress_live_with_feedback(sid: str, session: dict, agent, arg: str, *, snapshot_kwargs: bool) -> dict:
-    """Compress the live session; return the ``summarize_manual_compression`` dict.
+def _compress_live_with_feedback(sid: str, session: dict, agent, arg: str, *, snapshot_kwargs: bool) -> str:
+    """Compress the live session; return the user-facing feedback text.
 
     Shared by command.dispatch /compress and the slash mirror so every route shows
     "compressed N → M messages / ~X → ~Y tokens". ``snapshot_kwargs`` forwards the
     pre-read snapshot (approx_tokens/before_messages/history_version) to
     ``_compress_session_history``; the slash mirror passes only the raw arg. The raw
     arg goes through unparsed — the choke point parses ``here [N]`` / ``--keep N``.
-    CompressionLockHeld and other errors propagate to the caller, which finalizes
-    the deferred context-engine notification.
+    CompressionLockHeld is a clean no-op (its skip note is returned; the choke point
+    already discarded the deferred context-engine notification); other errors propagate
+    to the caller, which finalizes that notification.
     """
     from agent.conversation_compression import finalize_context_engine_compression_notification
-    from agent.manual_compression_feedback import summarize_manual_compression
+    from agent.manual_compression_feedback import describe_compression_lock_skip, summarize_manual_compression
     from agent.model_metadata import estimate_request_tokens_rough
-
     with session["history_lock"]:
         before_messages = list(session.get("history", []))
         history_version = int(session.get("history_version", 0))
@@ -293,38 +281,29 @@ def _compress_live_with_feedback(sid: str, session: dict, agent, arg: str, *, sn
     before_tokens = (
         estimate_request_tokens_rough(before_messages, system_prompt=sys_prompt, tools=tools) if before_messages else 0
     )
-    if snapshot_kwargs:
-        _compress_session_history(
-            session,
-            arg.strip() or None,
-            approx_tokens=before_tokens,
-            before_messages=before_messages,
-            history_version=history_version,
-        )
-    else:
-        _compress_session_history(session, arg)
+    try:
+        if snapshot_kwargs:
+            _compress_session_history(
+                session, arg.strip() or None, approx_tokens=before_tokens, before_messages=before_messages,
+                history_version=history_version)
+        else:
+            _compress_session_history(session, arg)
+    except CompressionLockHeld as e:
+        return describe_compression_lock_skip(e.holder)
     _sync_session_key_after_compress(sid, session)
     with session["history_lock"]:
         after_messages = list(session.get("history", []))
     after_tokens = (
         estimate_request_tokens_rough(
-            after_messages,
-            system_prompt=getattr(agent, "_cached_system_prompt", "") or sys_prompt,
-            tools=getattr(agent, "tools", None) or tools,
-        )
-        if after_messages
-        else 0
-    )
+            after_messages, system_prompt=getattr(agent, "_cached_system_prompt", "") or sys_prompt,
+            tools=getattr(agent, "tools", None) or tools)
+        if after_messages else 0)
     _emit("session.info", sid, _session_info(agent, session))
     fb = summarize_manual_compression(
-        before_messages,
-        after_messages,
-        before_tokens,
-        after_tokens,
-        compression_state=getattr(agent, "context_compressor", None),
-    )
+        before_messages, after_messages, before_tokens, after_tokens,
+        compression_state=getattr(agent, "context_compressor", None))
     finalize_context_engine_compression_notification(agent, committed=True)
-    return fb
+    return "\n".join(filter(None, [fb["headline"], fb["token_line"], fb.get("note")]))
 
 
 def _mirror_model(sid, session, agent, arg) -> str:
@@ -345,7 +324,6 @@ def _mirror_personality(sid, session, agent, arg) -> str:
         pname, new_prompt = _validate_personality(arg, _load_cfg())
         # Persist through the single owner so this surface never drifts from the others.
         from hermes_cli.personality import persist_personality
-
         persist_personality(pname)
         _apply_personality_to_session(sid, session, new_prompt, pname)
     return ""
@@ -360,29 +338,17 @@ def _mirror_prompt(sid, session, agent, arg) -> str:
 
 
 def _mirror_compress(sid, session, agent, arg) -> str:
-    if not agent:
-        return ""
-    try:
-        fb = _compress_live_with_feedback(sid, session, agent, arg, snapshot_kwargs=False)
-    except CompressionLockHeld as e:
-        from agent.manual_compression_feedback import describe_compression_lock_skip
+    return _compress_live_with_feedback(sid, session, agent, arg, snapshot_kwargs=False) if agent else ""
 
-        return describe_compression_lock_skip(e.holder)
-    lines = [fb["headline"], fb["token_line"]]
-    if fb.get("note"):
-        lines.append(fb["note"])
-    return "\n".join(lines)
+
+_FAST_TIERS = {"fast": "priority", "on": "priority", "normal": None, "off": None, "auto": "auto", "cold": "cold"}
 
 
 def _mirror_fast(sid, session, agent, arg) -> str:
     if agent:
         mode = arg.lower()
-        if mode in {"fast", "on"}:
-            agent.service_tier = "priority"
-        elif mode in {"normal", "off"}:
-            agent.service_tier = None
-        elif mode in {"auto", "cold"}:
-            agent.service_tier = mode
+        if mode in _FAST_TIERS:
+            agent.service_tier = _FAST_TIERS[mode]
         _emit("session.info", sid, _session_info(agent, session))
     return ""
 
@@ -395,7 +361,6 @@ def _mirror_reload_mcp(sid, session, agent, arg) -> str:
 
 def _mirror_stop(sid, session, agent, arg) -> str:
     from tools.process_registry import process_registry
-
     process_registry.kill_all()
     return ""
 
@@ -408,8 +373,7 @@ _SLASH_MIRRORS = {
     "compress": _mirror_compress,
     "fast": _mirror_fast,
     "reload-mcp": _mirror_reload_mcp,
-    "stop": _mirror_stop,
-}
+    "stop": _mirror_stop}
 
 
 def _compute_host_slash(sid: str, session: dict, name: str, command: str) -> tuple[str, str]:
@@ -426,13 +390,9 @@ def _compute_host_slash(sid: str, session: dict, name: str, command: str) -> tup
 
     def _on_late_ack(late: dict, _sid=sid) -> None:
         _adopt_late_compute_host_compress_ack(_sid, _late_session, late, route_name=route_name)
-
     try:
         ack = _send_compute_host_control(
-            sid,
-            route_name=route_name,
-            command=command,
-            wait=True,
+            sid, route_name=route_name, command=command, wait=True,
             **({"timeout": _compute_host_compress_wait_seconds(), "on_late_ack": _on_late_ack} if is_compress else {}),
         )
     except queue.Empty:
@@ -457,12 +417,10 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str:
         # /compact aliases /compress everywhere; the compute-host control forwards the
         # raw alias verbatim, so without this the child mirror silently no-ops.
         name = "compress"
-
     if _session_uses_compute_host(session) and name in _MUTATES_WHILE_RUNNING:
         return _compute_host_slash(sid, session, name, command)[1]
     if name in _MUTATES_WHILE_RUNNING and session.get("running"):
         return f"session busy — /interrupt the current turn before running /{name}"
-
     mirror = _SLASH_MIRRORS.get(name)
     if mirror is None:
         return ""
@@ -471,7 +429,6 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str:
     except Exception as e:
         if name == "compress" and agent:
             from agent.conversation_compression import finalize_context_engine_compression_notification
-
             finalize_context_engine_compression_notification(agent, committed=False)
         return f"live session sync failed: {e}"
 
diff --git a/tui_gateway/methods_tools.py b/tui_gateway/methods_tools.py
index 6002256861..5657537088 100644
--- a/tui_gateway/methods_tools.py
+++ b/tui_gateway/methods_tools.py
@@ -39,7 +39,6 @@ def _profile_scoped_rpc(fail_code: int, *, required=(), catch_resolve: bool = Tr
                 try:
                     from hermes_cli.profiles import get_profile_dir
                     from hermes_constants import set_hermes_home_override
-
                     profile_dir = get_profile_dir(profile)
                     if not profile_dir or not profile_dir.is_dir():
                         return _err(rid, 4064, f"profile '{profile}' not found")
@@ -54,10 +53,8 @@ def _profile_scoped_rpc(fail_code: int, *, required=(), catch_resolve: bool = Tr
                 return _err(rid, fail_code, f"{prefix}{e}")
             finally:
                 _mcp_reset_profile(token)
-
         handler.__doc__ = body.__doc__
         return handler
-
     return deco
 
 
@@ -66,6 +63,24 @@ def _guarded(fail_code: int, prefix: str = ""):
     return _profile_scoped_rpc(fail_code, prefix=prefix, scoped=False)
 
 
+def _live_session_guarded(fail_code: int):
+    """Resolve the session via ``_sess`` (waits for the agent build) and call
+    ``body(rid, params, session)``; body exceptions → ``fail_code``."""
+
+    def deco(body):
+        def handler(rid, params: dict) -> dict:
+            session, err = _sess(params, rid)
+            if err:
+                return err
+            try:
+                return body(rid, params, session)
+            except Exception as e:
+                return _err(rid, fail_code, str(e))
+        handler.__doc__ = body.__doc__
+        return handler
+    return deco
+
+
 def _stripped(v) -> bool:
     return bool(str(v or "").strip())
 
@@ -86,7 +101,6 @@ def _mcp_server_scoped(body):
 def _mcp_named_server(rid, params):
     """(name, servers, None) for a configured server, else (name, servers, 4064 error)."""
     from hermes_cli.mcp_config import _get_mcp_servers
-
     name = str(params.get("name") or "").strip()
     servers = _get_mcp_servers()
     err = None if name in servers else _err(rid, 4064, f"server '{name}' not found")
@@ -112,7 +126,6 @@ def _session_key_or_err(rid, session):
 def _user_turn_indices(session):
     """(history, indices of user-originated turns) minus ephemeral scaffolding. Call under history_lock."""
     from agent.context_compressor import user_originated_turn_view
-
     history = _history_without_ephemeral_scaffolding(session.get("history", []))
     return history, [i for i, m in enumerate(history) if user_originated_turn_view(m) is not None]
 
@@ -126,7 +139,6 @@ def _capture_run_kwargs(timeout: int) -> dict:
     text, UTF-8 + lossy decode (non-UTF-8 child output must not crash the gateway thread
     on locale-mismatched Windows), no stdin, no console flash under the desktop parent."""
     from hermes_cli._subprocess_compat import windows_hide_flags
-
     return dict(
         capture_output=True,
         text=True,
@@ -134,13 +146,11 @@ def _capture_run_kwargs(timeout: int) -> dict:
         errors="replace",
         timeout=timeout,
         stdin=subprocess.DEVNULL,
-        creationflags=windows_hide_flags(),
-    )
+        creationflags=windows_hide_flags())
 
 
 def _toolset_rows(params: dict, *, with_tools: bool) -> list[dict]:
     from toolsets import get_all_toolsets, get_toolset_info
-
     session = _sessions.get(params.get("session_id", ""))
     enabled = (
         set(getattr(session["agent"], "enabled_toolsets", []) or []) if session else set(_load_enabled_toolsets() or [])
@@ -154,8 +164,7 @@ def _toolset_rows(params: dict, *, with_tools: bool) -> list[dict]:
             "name": name,
             "description": info["description"],
             "tool_count": info["tool_count"],
-            "enabled": name in enabled if enabled else True,
-        }
+            "enabled": name in enabled if enabled else True}
         if with_tools:
             row["tools"] = info["resolved_tools"]
         items.append(row)
@@ -170,7 +179,6 @@ def _(rid, params: dict) -> dict:
     """Host battery for the status bar. Always resolves; ``available: false`` = no battery or read failed."""
     try:
         from agent.battery import battery_category, read_battery
-
         batt = read_battery()
         return _ok(
             rid,
@@ -178,9 +186,7 @@ def _(rid, params: dict) -> dict:
                 "available": batt.available,
                 "percent": batt.percent,
                 "plugged": batt.plugged,
-                "category": battery_category(batt),
-            },
-        )
+                "category": battery_category(batt)})
     except Exception:
         return _ok(rid, {"available": False, "percent": None, "plugged": None, "category": "dim"})
 
@@ -189,47 +195,34 @@ def _(rid, params: dict) -> dict:
 @_guarded(5010)
 def _(rid, params: dict) -> dict:
     from tools.process_registry import process_registry
-
     return _ok(rid, {"killed": process_registry.kill_all()})
 
 
 @method("process.list")
-def _(rid, params: dict) -> dict:
+@_live_session_guarded(5010)
+def _(rid, params: dict, session) -> dict:
     """Session-scoped view of the background process registry (desktop status stack)."""
-    session, err = _sess(params, rid)
-    if err:
-        return err
-    try:
-        return _ok(rid, {"processes": _session_processes(session)})
-    except Exception as e:
-        return _err(rid, 5010, str(e))
+    return _ok(rid, {"processes": _session_processes(session)})
 
 
 @method("process.kill")
-def _(rid, params: dict) -> dict:
+@_live_session_guarded(5010)
+def _(rid, params: dict, session) -> dict:
     """Kill ONE background process, scoped to the caller's session (unlike process.stop's kill_all)."""
-    session, err = _sess(params, rid)
-    if err:
-        return err
     proc_id = str(params.get("process_id") or "")
     if not proc_id:
         return _err(rid, 4012, "process_id required")
-    try:
-        from tools.process_registry import process_registry
-
-        proc = process_registry.get(proc_id)
-        if proc is None or str(getattr(proc, "session_key", "") or "") != str(session.get("session_key") or ""):
-            return _err(rid, 4044, f"no such process: {proc_id}")
-        return _ok(rid, process_registry.kill_process(proc_id))
-    except Exception as e:
-        return _err(rid, 5010, str(e))
+    from tools.process_registry import process_registry
+    proc = process_registry.get(proc_id)
+    if proc is None or str(getattr(proc, "session_key", "") or "") != str(session.get("session_key") or ""):
+        return _err(rid, 4044, f"no such process: {proc_id}")
+    return _ok(rid, process_registry.kill_process(proc_id))
 
 
 def _mcp_reload_confirm_required() -> bool:
     """``approvals.mcp_reload_confirm`` from disk config; True (safe) on any failure."""
     try:
         from hermes_cli.config import load_config
-
         cfg = load_config()
         approvals = cfg.get("approvals") if isinstance(cfg, dict) else None
         return bool(approvals.get("mcp_reload_confirm", True)) if isinstance(approvals, dict) else True
@@ -248,19 +241,15 @@ def _(rid, params: dict) -> dict:
         message = (
             "⚠️  /reload-mcp invalidates the prompt cache (next message re-sends full input tokens). "
             "Reply `/reload-mcp now` to proceed, or `/reload-mcp always` to proceed and "
-            "silence this prompt permanently."
-        )
+            "silence this prompt permanently.")
         return _ok(rid, {"status": "confirm_required", "message": message})
-
     if session and _session_uses_compute_host(session):
         try:
             ack = _get_compute_host_supervisor().reload_mcp(
-                str(params.get("session_id") or ""), request_id=f"reload-mcp-{rid}"
-            )
+                str(params.get("session_id") or ""), request_id=f"reload-mcp-{rid}")
         except Exception as exc:
             return _err(rid, 5019, f"compute-host reload_mcp failed: {exc}")
         return _ok(rid, {"status": "reloaded", "turn_isolation": True, "host_ack": ack})
-
     from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools, reprobe_tool_availability
 
     def _refresh_session_agent() -> None:
@@ -279,7 +268,6 @@ def _(rid, params: dict) -> dict:
         except Exception as _exc:
             logger.warning("Failed to refresh cached agent tools after /reload-mcp: %s", _exc)
         _emit("session.info", params.get("session_id", ""), _session_info(agent, session))
-
     global _mcp_reload_gen, _mcp_reload_loaded_rev
 
     # Revision the CALLER wants loaded (the mcp_rev its poll observed); empty on
@@ -292,7 +280,6 @@ def _(rid, params: dict) -> dict:
         down mid-rebuild. Config can change WHILE discover connects: re-hash after
         discovery and repeat until stable so the marked generation matches what loaded."""
         global _mcp_reload_gen, _mcp_reload_loaded_rev
-
         loaded = _compute_mcp_rev()
         for _ in range(_MCP_RELOAD_MAX_PASSES):
             shutdown_mcp_servers()
@@ -302,7 +289,6 @@ def _(rid, params: dict) -> dict:
             if after == loaded:
                 break
             loaded = after
-
         _refresh_session_agent()
         _mcp_reload_loaded_rev = loaded
         _mcp_reload_gen += 1
@@ -317,22 +303,17 @@ def _(rid, params: dict) -> dict:
             _do_full_reload()
         finally:
             _mcp_reload_lock.release()
-
         return _finish_reload(rid, params, coalesced=False)
-
     gen_before = _mcp_reload_gen
-
     with _mcp_reload_lock:
         leader_completed = _mcp_reload_gen > gen_before
         rev_satisfied = not req_rev or req_rev == _mcp_reload_loaded_rev
-
         if leader_completed and rev_satisfied:
             _refresh_session_agent()
             coalesced = True
         else:
             _do_full_reload()
             coalesced = False
-
     return _finish_reload(rid, params, coalesced=coalesced)
 
 
@@ -342,7 +323,6 @@ def _(rid, params: dict) -> dict:
     """Re-read ``~/.hermes/.env`` (classic CLI ``/reload`` parity). Already-built agents
     keep their credential pool / provider routing; ``/new`` resolves fresh."""
     from hermes_cli.config import reload_env
-
     return _ok(rid, {"updated": int(reload_env())})
 
 
@@ -354,7 +334,6 @@ def _(rid, params: dict) -> dict:
 def _(rid, params: dict) -> dict:
     """Registry-backed slash metadata for the TUI — categorized, no aliases."""
     from hermes_cli.commands import COMMAND_REGISTRY, SUBCOMMANDS, _build_description, command_desktop_meta
-
     all_pairs: list[list[str]] = []
     canon: dict[str, str] = {}
     commands: dict[str, dict[str, str | None]] = {}
@@ -371,7 +350,6 @@ def _(rid, params: dict) -> dict:
         canon[key.lower()] = key
         all_pairs.append([key, desc])
         rows.append([key, desc])
-
     for cmd in COMMAND_REGISTRY:
         meta = command_desktop_meta(cmd)
         commands[f"/{cmd.name}"] = dict(meta)
@@ -383,12 +361,10 @@ def _(rid, params: dict) -> dict:
         add(c, _build_description(cmd), bucket(cmd.category))
         for a in cmd.aliases:
             canon[f"/{a}".lower()] = c
-
     for name, desc, cat in _TUI_EXTRA:
         # Registry command/alias wins over a colliding TUI extra (e.g. /compact, /sessions).
         if name.lower() not in canon:
             add(name, desc, bucket(cat))
-
     warning = ""
     try:
         qcmds = _load_cfg().get("quick_commands", {}) or {}
@@ -405,10 +381,8 @@ def _(rid, params: dict) -> dict:
                 add(f"/{qname}", _clip(str(qc.get("description") or default_desc)), rows)
     except Exception as e:
         warning = f"quick_commands discovery unavailable: {e}"
-
     try:
         from hermes_cli.plugins import get_plugin_commands
-
         plugin_cmds = get_plugin_commands() or {}
         if plugin_cmds:
             rows = bucket("Plugin commands")
@@ -427,7 +401,6 @@ def _(rid, params: dict) -> dict:
     except Exception as e:
         if not warning:
             warning = f"plugin command discovery unavailable: {e}"
-
     skill_count = 0
     skills: dict[str, dict] = {}
     try:
@@ -435,7 +408,6 @@ def _(rid, params: dict) -> dict:
 
         # Usage + origin ride along (not a second RPC): every catalog consumer also ranks it.
         usage, origin_of = _skill_usage_lookup()
-
         for k, info in sorted(scan_skill_commands().items()):
             all_pairs.append([k, _clip(str(info.get("description", "Skill")))])
             name = str(info.get("name") or k.lstrip("/"))
@@ -443,7 +415,6 @@ def _(rid, params: dict) -> dict:
             skill_count += 1
     except Exception as e:
         warning = f"skill discovery unavailable: {e}"
-
     payload = {
         "pairs": all_pairs,
         "sub": {k: v[:] for k, v in SUBCOMMANDS.items()},
@@ -452,8 +423,7 @@ def _(rid, params: dict) -> dict:
         "categories": [{"name": cat, "pairs": cat_map[cat]} for cat in cat_order],
         "skills": skills,
         "skill_count": skill_count,
-        "warning": warning,
-    }
+        "warning": warning}
     return _ok(rid, payload)
 
 
@@ -472,8 +442,7 @@ def _(rid, params: dict) -> dict:
             cwd=os.getcwd(),
             # Can drive the agent → needs provider credentials; tier-1 secrets still stripped.
             env=hermes_subprocess_env(inherit_credentials=True),
-            **_capture_run_kwargs(min(int(params.get("timeout", 240)), 600)),
-        )
+            **_capture_run_kwargs(min(int(params.get("timeout", 240)), 600)))
         parts = [r.stdout or "", r.stderr or ""]
         out = "\n".join(p for p in parts if p).strip() or "(no output)"
         return _ok(rid, {"blocked": False, "code": r.returncode, "output": out[:48_000]})
@@ -487,7 +456,6 @@ def _(rid, params: dict) -> dict:
 @_guarded(5012)
 def _(rid, params: dict) -> dict:
     from hermes_cli.commands import resolve_command
-
     r = resolve_command(params.get("name", ""))
     if r:
         return _ok(rid, {"canonical": r.name, "description": r.description, "category": r.category})
@@ -506,13 +474,11 @@ def _dispatch_quick(rid, params, session, name, arg):
     if qc.get("type") == "exec":
         # Sanitized env: the TUI server process holds every API key in os.environ.
         from tools.environments.local import build_subprocess_env
-
         sanitized_env = build_subprocess_env()
         r = subprocess.run(qc.get("command", ""), shell=True, env=sanitized_env, **_capture_run_kwargs(30))
         output = ((r.stdout or "") + ("\n" if r.stdout and r.stderr else "") + (r.stderr or "")).strip()[:4000]
         if output:
             from agent.redact import redact_sensitive_text
-
             output = redact_sensitive_text(output)
         if r.returncode != 0:
             return _err(rid, 4018, output or f"quick command failed with exit code {r.returncode}")
@@ -525,12 +491,16 @@ def _dispatch_quick(rid, params, session, name, arg):
 def _plugin_command_handler(name: str):
     try:
         from hermes_cli.plugins import get_plugin_command_handler
-
         return get_plugin_command_handler(name)
     except Exception:
         return None
 
 
+def _run_plugin_command(handler, arg: str) -> str:
+    from hermes_cli.plugins import resolve_plugin_command_result
+    return str(resolve_plugin_command_result(handler(arg)) or "")
+
+
 def _is_profile_skill_command(session: dict, base: str) -> bool:
     """True when ``/base`` is a skill command of the session's profile. HERMES_HOME is bound
     to that profile so get_skill_commands() sees its skills.external_dirs: dispatch() runs on
@@ -538,7 +508,6 @@ def _is_profile_skill_command(session: dict, base: str) -> bool:
     try:
         from agent.skill_commands import get_skill_commands
         from hermes_constants import reset_hermes_home_override, set_hermes_home_override
-
         profile_home = session.get("profile_home")
         token = set_hermes_home_override(profile_home) if profile_home else None
         try:
@@ -553,13 +522,8 @@ def _is_profile_skill_command(session: dict, base: str) -> bool:
 def _dispatch_plugin(rid, params, session, name, arg):
     handler = _plugin_command_handler(name)
     if handler:
-        try:
-            from hermes_cli.plugins import resolve_plugin_command_result
-
-            result = resolve_plugin_command_result(handler(arg))
-            return _ok(rid, {"type": "plugin", "output": str(result or "")})
-        except Exception:
-            pass
+        with contextlib.suppress(Exception):
+            return _ok(rid, {"type": "plugin", "output": _run_plugin_command(handler, arg)})
     return None
 
 
@@ -568,7 +532,6 @@ def _bundle_key_for(name: str):
     try:
         from agent.skill_bundles import resolve_bundle_command_key
         from hermes_cli.commands import resolve_command
-
         return resolve_bundle_command_key(name) if resolve_command(name) is None else None
     except Exception:
         return None
@@ -579,19 +542,16 @@ def _dispatch_bundle(rid, params, session, name, arg):
     if bundle_key is None:
         return None
     from agent.skill_bundles import build_bundle_invocation_message, get_skill_bundles
-
     try:
         bundle_result = build_bundle_invocation_message(
             bundle_key,
             arg,
             task_id=session.get("session_key", "") if session else "",
-            platform=_resolve_session_platform(),
-        )
+            platform=_resolve_session_platform())
     except Exception as exc:
         return _err(rid, 4018, f"bundle dispatch failed: {exc}")
     if not bundle_result:
         return _err(rid, 4018, f"failed to load bundle: {bundle_key}")
-
     msg, loaded_names, missing = bundle_result
     bundle_name = get_skill_bundles().get(bundle_key, {}).get("name", bundle_key.lstrip("/"))
     notice = f"⚡ Loading bundle: {bundle_name} ({len(loaded_names)} skills)"
@@ -604,7 +564,6 @@ def _dispatch_bundle(rid, params, session, name, arg):
 def _dispatch_skill(rid, params, session, name, arg):
     try:
         from agent.skill_commands import scan_skill_commands, build_skill_invocation_message
-
         cmds = scan_skill_commands()
         key = f"/{name}"
         if key in cmds:
@@ -631,21 +590,18 @@ def _cmd_queue(rid, params, session, name, arg):
 def _cmd_learn(rid, params, session, name, arg):
     # Submitted as a normal turn; the live agent gathers sources and authors the skill via skill_manage.
     from agent.learn_prompt import build_learn_prompt
-
     return _ok(rid, {"type": "send", "message": build_learn_prompt(arg)})
 
 
 def _cmd_plan(rid, params, session, name, arg):
     # Normal turn (as /learn); the agent saves the plan under .hermes/plans/ via write_file.
     from agent.plan_prompt import build_plan_prompt
-
     return _ok(rid, {"type": "send", "message": build_plan_prompt(arg)})
 
 
 def _cmd_init(rid, params, session, name, arg):
     # Generate-or-update AGENTS.md as a normal turn (as /learn).
     from hermes_cli.init_command import build_init_prompt_for_cwd
-
     return _ok(rid, {"type": "send", "message": build_init_prompt_for_cwd(extra=arg)})
 
 
@@ -654,7 +610,6 @@ def _cmd_moa(rid, params, session, name, arg):
     # switching goes through the model picker (MoA presets = virtual "Mixture of Agents" provider).
     try:
         from hermes_cli.moa_config import moa_usage, normalize_moa_config
-
         if not arg:
             return _err(rid, 4004, moa_usage())
         if not session:
@@ -667,8 +622,7 @@ def _cmd_moa(rid, params, session, name, arg):
         session["moa_one_shot_restore"] = {
             "override": session.get("model_override"),
             "model": getattr(agent, "model", None) if agent else None,
-            "provider": getattr(agent, "provider", None) if agent else None,
-        }
+            "provider": getattr(agent, "provider", None) if agent else None}
         if agent is not None:
             try:
                 _apply_model_switch(
@@ -689,8 +643,7 @@ def _cmd_moa(rid, params, session, name, arg):
                 "model": preset,
                 "base_url": "moa://local",
                 "api_key": "moa-virtual-provider",
-                "api_mode": "chat_completions",
-            }
+                "api_mode": "chat_completions"}
         notice = f"MoA one-shot queued with preset {preset}; previous model will be restored after this turn."
         return _ok(rid, {"type": "send", "notice": notice, "message": arg})
     except Exception as exc:
@@ -700,23 +653,21 @@ def _cmd_moa(rid, params, session, name, arg):
 def _cmd_focus(rid, params, session, name, arg):
     # Display-only; routed through the config.set branch Ink uses so both surfaces share one state machine.
     from hermes_cli.focus_view import format_focus_status, format_focus_toggle_message, resolve_focus_arg
-
-    _display_focus = _load_cfg().get("display")
-    _d_focus: dict = _display_focus if isinstance(_display_focus, dict) else {}
-    _cur_focus = bool(_d_focus.get("focus_view", False))
-    _action, _target = resolve_focus_arg(arg, _cur_focus)
-    if _action == "usage":
+    display = _load_cfg().get("display")
+    display = display if isinstance(display, dict) else {}
+    cur = bool(display.get("focus_view", False))
+    action, target = resolve_focus_arg(arg, cur)
+    if action == "usage":
         return _err(rid, 4004, "usage: /focus [on|off|status]")
-    if _action == "status":
-        _saved = _d_focus.get("focus_saved_tool_progress") or _load_tool_progress_mode()
-        return _ok(rid, {"type": "exec", "output": format_focus_status(_cur_focus, _saved)})
-    _res = _methods["config.set"](
-        rid, {"key": "focus", "value": "on" if _target else "off", "session_id": params.get("session_id", "")}
+    if action == "status":
+        saved = display.get("focus_saved_tool_progress") or _load_tool_progress_mode()
+        return _ok(rid, {"type": "exec", "output": format_focus_status(cur, saved)})
+    res = _methods["config.set"](
+        rid, {"key": "focus", "value": "on" if target else "off", "session_id": params.get("session_id", "")}
     )
-    if "error" in _res:
-        return _res
-    _payload = _res.get("result") or {}
-    output = format_focus_toggle_message(bool(_target), _payload.get("tool_progress") or "all")
+    if "error" in res:
+        return res
+    output = format_focus_toggle_message(bool(target), (res.get("result") or {}).get("tool_progress") or "all")
     return _ok(rid, {"type": "exec", "output": output})
 
 
@@ -726,7 +677,6 @@ def _cmd_retry(rid, params, session, name, arg):
     if busy := _busy_error(rid, session, "retry"):
         return busy
     from agent.context_compressor import history_before_user_originated_turn, retryable_user_text
-
     with session["history_lock"]:
         if busy := _busy_error(rid, session, "retry"):
             return busy
@@ -742,8 +692,7 @@ def _cmd_retry(rid, params, session, name, arg):
             return _err(rid, 4018, str(exc))
         try:
             _active, durable_live_view, _rewound_count = _rewind_active_session_history(
-                session, len(user_indices) - 1, require_retryable=True
-            )
+                session, len(user_indices) - 1, require_retryable=True)
         except ValueError as exc:
             return _err(rid, 4018, str(exc))
         except Exception as exc:
@@ -768,21 +717,18 @@ def _cmd_steer(rid, params, session, name, arg):
 
 
 def _cmd_goal(rid, params, session, name, arg):
-    if not session:
-        return _err(rid, 4001, "no active session")
+    sid_key, err = _session_key_or_err(rid, session)
+    if err:
+        return err
     try:
         from hermes_cli.goals import GoalManager
     except Exception as exc:
         return _err(rid, 5030, f"goals unavailable: {exc}")
-    sid_key, err = _session_key_or_err(rid, session)
-    if err:
-        return err
     try:
         max_turns = int((_load_cfg().get("goals") or {}).get("max_turns", 20) or 20)
     except Exception:
         max_turns = 20
     mgr = GoalManager(session_id=sid_key, default_max_turns=max_turns)
-
     lower = arg.strip().lower()
     if not arg.strip() or lower == "status":
         return _ok(rid, {"type": "exec", "output": mgr.status_line()})
@@ -814,33 +760,28 @@ def _cmd_goal(rid, params, session, name, arg):
     notice = (
         f"⊙ Goal set ({state.max_turns}-turn budget): {state.goal}\n"
         "I'll keep working until the goal is done, you pause/clear it, or the budget is exhausted.\n"
-        "Controls: /goal status · /goal pause · /goal resume · /goal clear"
-    )
+        "Controls: /goal status · /goal pause · /goal resume · /goal clear")
     return _ok(rid, {"type": "send", "notice": notice, "message": state.goal})
 
 
 def _cmd_loop(rid, params, session, name, arg):
     # Recurring in-session wakeups; the notification poller fires due ones while the session is idle.
-    if not session:
-        return _err(rid, 4001, "no active session")
+    sid_key, err = _session_key_or_err(rid, session)
+    if err:
+        return err
     try:
         from hermes_cli.loops import LoopManager, dispatch_loop_command
     except Exception as exc:
         return _err(rid, 5030, f"loops unavailable: {exc}")
-    sid_key, err = _session_key_or_err(rid, session)
-    if err:
-        return err
     result = dispatch_loop_command(LoopManager(session_id=sid_key), arg)
     output = result.get("output") or ""
     if result.get("created"):
         with contextlib.suppress(Exception):
             from hermes_cli.loops import goal_blocks_loop_tick
-
             if goal_blocks_loop_tick(sid_key):
                 output += (
                     "\nNote: an active /goal is driving this session — loop "
-                    "wakeups defer until the goal finishes, pauses, or parks."
-                )
+                    "wakeups defer until the goal finishes, pauses, or parks.")
     return _ok(rid, {"type": "exec", "output": output})
 
 
@@ -862,7 +803,6 @@ def _cmd_undo(rid, params, session, name, arg):
             return _err(rid, 4004, f"undo: invalid count {arg_str!r} — use /undo or /undo N")
     n = max(n, 1)
     from agent.message_content import flatten_message_text
-
     with session["history_lock"]:
         if busy := _busy_error(rid, session, "undo"):
             return busy
@@ -912,7 +852,6 @@ def _cmd_compress(rid, params, session, name, arg):
     if busy := _busy_error(rid, session, "compress"):
         return busy
     from agent.conversation_compression import finalize_context_engine_compression_notification
-
     sid = params.get("session_id", "")
     if _session_uses_compute_host(session):
         status, text = _compute_host_slash(sid, session, "compress", f"/{name}" + (f" {arg}" if arg else ""))
@@ -921,15 +860,8 @@ def _cmd_compress(rid, params, session, name, arg):
         payload = {"type": "exec", "status": "pending", "output": text} if status == "pending" else {"type": "exec", "output": text}
         return _ok(rid, payload)
     try:
-        summary = _compress_live_with_feedback(sid, session, session["agent"], arg, snapshot_kwargs=True)
-        output = "\n".join(filter(None, [summary["headline"], summary["token_line"], summary.get("note")]))
+        output = _compress_live_with_feedback(sid, session, session["agent"], arg, snapshot_kwargs=True)
         return _ok(rid, {"type": "exec", "output": output})
-    except CompressionLockHeld as e:
-        # Clean no-op (parity with the slash mirror / session.compress), never "compress failed";
-        # _compress_session_history already discarded the deferred context-engine notification.
-        from agent.manual_compression_feedback import describe_compression_lock_skip
-
-        return _ok(rid, {"type": "exec", "output": describe_compression_lock_skip(e.holder)})
     except Exception as exc:
         finalize_context_engine_compression_notification(session["agent"], committed=False)
         return _err(rid, 5009, f"compress failed: {exc}")
@@ -940,8 +872,7 @@ _SLASH_BUILTINS = {
     "queue": _cmd_queue, "q": _cmd_queue, "learn": _cmd_learn, "plan": _cmd_plan, "init": _cmd_init,
     "moa": _cmd_moa, "focus": _cmd_focus, "retry": _cmd_retry, "steer": _cmd_steer, "goal": _cmd_goal,
     "loop": _cmd_loop, "undo": _cmd_undo, "snapshot": _cmd_snapshot, "snap": _cmd_snapshot,
-    "compress": _cmd_compress, "compact": _cmd_compress,
-}
+    "compress": _cmd_compress, "compact": _cmd_compress}
 
 
 @method("command.dispatch")
@@ -968,49 +899,37 @@ def _(rid, params: dict) -> dict:
     session, err = _sess_nowait(params, rid)
     if err:
         return err
-
     cmd = params.get("command", "").strip()
     if not cmd:
         return _err(rid, 4004, "empty command")
 
     # Skill/bundle and _PENDING_INPUT_COMMANDS must NOT reach the slash worker. Plugin
     # commands also bypass it but return normal slash.exec output (TUI keeps the pager path).
-    _cmd_text = cmd.lstrip("/") if cmd.startswith("/") else cmd
-    _cmd_parts = _cmd_text.split(maxsplit=1)
-    _cmd_base = (_cmd_parts[0] if _cmd_parts else "").lower()
-    _cmd_arg = _cmd_parts[1] if len(_cmd_parts) > 1 else ""
+    parts = cmd.lstrip("/").split(maxsplit=1)
+    base = (parts[0] if parts else "").lower()
+    arg = parts[1] if len(parts) > 1 else ""
     sid = params.get("session_id", "")
-
-    live_output = _live_slash_command_output(sid, session, _cmd_base, _cmd_arg)
+    live_output = _live_slash_command_output(sid, session, base, arg)
     if live_output is not None:
         return _ok(rid, {"output": live_output or "(no output)"})
-
-    if _cmd_base in _PENDING_INPUT_COMMANDS:
+    if base in _PENDING_INPUT_COMMANDS:
         # Route straight to command.dispatch: some clients fail the error-then-retry fallback ("empty command").
-        return _methods["command.dispatch"](rid, {"name": _cmd_base, "arg": _cmd_arg, "session_id": sid})
-
-    if _cmd_base in _WORKER_BLOCKED_COMMANDS:
-        subcommand = _cmd_arg.split(maxsplit=1)[0].lower() if _cmd_arg else ""
+        return _methods["command.dispatch"](rid, {"name": base, "arg": arg, "session_id": sid})
+    if base in _WORKER_BLOCKED_COMMANDS:
+        subcommand = arg.split(maxsplit=1)[0].lower() if arg else ""
         if subcommand in {"restore", "rewind"}:
             return _err(rid, 4018, "snapshot restore mutates live config/state; use command.dispatch for /snapshot restore")
-
-    _bundle_key = _bundle_key_for(_cmd_base)
-    if _bundle_key is not None:
-        return _methods["command.dispatch"](rid, {"name": _bundle_key.lstrip("/"), "arg": _cmd_arg, "session_id": sid})
-
-    if _is_profile_skill_command(session, _cmd_base):
-        return _err(rid, 4018, f"skill command: use command.dispatch for /{_cmd_base}")
-
-    plugin_handler = _plugin_command_handler(_cmd_base) if _cmd_base else None
+    bundle_key = _bundle_key_for(base)
+    if bundle_key is not None:
+        return _methods["command.dispatch"](rid, {"name": bundle_key.lstrip("/"), "arg": arg, "session_id": sid})
+    if _is_profile_skill_command(session, base):
+        return _err(rid, 4018, f"skill command: use command.dispatch for /{base}")
+    plugin_handler = _plugin_command_handler(base) if base else None
     if plugin_handler:
         try:
-            from hermes_cli.plugins import resolve_plugin_command_result
-
-            result = resolve_plugin_command_result(plugin_handler(_cmd_arg))
-            return _ok(rid, {"output": str(result or "(no output)")})
+            return _ok(rid, {"output": _run_plugin_command(plugin_handler, arg) or "(no output)"})
         except Exception as e:
             return _ok(rid, {"output": f"Plugin command error: {e}"})
-
     worker = session.get("slash_worker")
     if not worker:
         # slash.exec runs on the RPC pool: two concurrent commands could both see
@@ -1025,12 +944,10 @@ def _(rid, params: dict) -> dict:
                     worker = _SlashWorker(
                         session["session_key"],
                         getattr(session.get("agent"), "model", _resolve_model()),
-                        profile_home=session.get("profile_home"),
-                    )
+                        profile_home=session.get("profile_home"))
                     _attach_worker(sid, session, worker)
                 except Exception as e:
                     return _err(rid, 5030, f"slash worker start failed: {e}")
-
     try:
         output = worker.run(cmd)
         warning = _mirror_slash_side_effects(sid, session, cmd)
@@ -1063,31 +980,21 @@ def _(rid, params: dict) -> dict:
 
 
 @method("rollback.list")
-def _(rid, params: dict) -> dict:
-    session, err = _sess(params, rid)
-    if err:
-        return err
-    try:
-
-        def go(mgr, cwd):
-            if not mgr.enabled:
-                return _ok(rid, {"enabled": False, "checkpoints": []})
-            rows = [
-                {"hash": c.get("hash", ""), "timestamp": c.get("timestamp", ""), "message": c.get("message", "")}
-                for c in mgr.list_checkpoints(cwd)
-            ]
-            return _ok(rid, {"enabled": True, "checkpoints": rows})
-
-        return _with_checkpoints(session, go)
-    except Exception as e:
-        return _err(rid, 5020, str(e))
+@_live_session_guarded(5020)
+def _(rid, params: dict, session) -> dict:
+    def go(mgr, cwd):
+        if not mgr.enabled:
+            return _ok(rid, {"enabled": False, "checkpoints": []})
+        rows = [
+            {"hash": c.get("hash", ""), "timestamp": c.get("timestamp", ""), "message": c.get("message", "")}
+            for c in mgr.list_checkpoints(cwd)]
+        return _ok(rid, {"enabled": True, "checkpoints": rows})
+    return _with_checkpoints(session, go)
 
 
 @method("rollback.restore")
-def _(rid, params: dict) -> dict:
-    session, err = _sess(params, rid)
-    if err:
-        return err
+@_live_session_guarded(5021)
+def _(rid, params: dict, session) -> dict:
     target = params.get("hash", "")
     file_path = params.get("file_path", "")
     if not target:
@@ -1096,46 +1003,37 @@ def _(rid, params: dict) -> dict:
     # would drop the agent's output or clobber it). File-scoped only touches disk.
     if not file_path and session.get("running"):
         return _err(rid, 4009, "session busy — /interrupt the current turn before full rollback.restore")
-    try:
 
-        def go(mgr, cwd):
-            resolved = _resolve_checkpoint_hash(mgr, cwd, target)
-            result = mgr.restore(cwd, resolved, file_path=file_path or None)
-            if result.get("success") and not file_path:
-                removed = 0
-                with session["history_lock"]:
-                    _history, user_indices = _user_turn_indices(session)
-                    if user_indices:
-                        try:
-                            _active, _live_view, removed = _rewind_active_session_history(session, len(user_indices) - 1)
-                        except Exception as exc:
-                            raise RuntimeError(f"checkpoint restored, but session history rewind failed: {exc}") from exc
-                result["history_removed"] = removed
-            return result
-
-        return _ok(rid, _with_checkpoints(session, go))
-    except Exception as e:
-        return _err(rid, 5021, str(e))
+    def go(mgr, cwd):
+        resolved = _resolve_checkpoint_hash(mgr, cwd, target)
+        result = mgr.restore(cwd, resolved, file_path=file_path or None)
+        if result.get("success") and not file_path:
+            removed = 0
+            with session["history_lock"]:
+                _history, user_indices = _user_turn_indices(session)
+                if user_indices:
+                    try:
+                        _active, _live_view, removed = _rewind_active_session_history(session, len(user_indices) - 1)
+                    except Exception as exc:
+                        raise RuntimeError(f"checkpoint restored, but session history rewind failed: {exc}") from exc
+            result["history_removed"] = removed
+        return result
+    return _ok(rid, _with_checkpoints(session, go))
 
 
 @method("rollback.diff")
-def _(rid, params: dict) -> dict:
-    session, err = _sess(params, rid)
-    if err:
-        return err
+@_live_session_guarded(5022)
+def _(rid, params: dict, session) -> dict:
     target = params.get("hash", "")
     if not target:
         return _err(rid, 4014, "hash required")
-    try:
-        r = _with_checkpoints(session, lambda mgr, cwd: mgr.diff(cwd, _resolve_checkpoint_hash(mgr, cwd, target)))
-        raw = r.get("diff", "")[:4000]
-        payload = {"stat": r.get("stat", ""), "diff": raw}
-        rendered = render_diff(raw, session.get("cols", 80))
-        if rendered:
-            payload["rendered"] = rendered
-        return _ok(rid, payload)
-    except Exception as e:
-        return _err(rid, 5022, str(e))
+    r = _with_checkpoints(session, lambda mgr, cwd: mgr.diff(cwd, _resolve_checkpoint_hash(mgr, cwd, target)))
+    raw = r.get("diff", "")[:4000]
+    payload = {"stat": r.get("stat", ""), "diff": raw}
+    rendered = render_diff(raw, session.get("cols", 80))
+    if rendered:
+        payload["rendered"] = rendered
+    return _ok(rid, payload)
 
 
 @method("browser.manage")
@@ -1155,11 +1053,9 @@ def _(rid, params: dict) -> dict:
 @_guarded(5032)
 def _(rid, params: dict) -> dict:
     from hermes_cli.plugins import get_plugin_manager
-
     rows = [
         {"name": n, "version": getattr(i, "version", "?"), "enabled": getattr(i, "enabled", True)}
-        for n, i in get_plugin_manager()._plugins.items()
-    ]
+        for n, i in get_plugin_manager()._plugins.items()]
     return _ok(rid, {"plugins": rows})
 
 
@@ -1169,15 +1065,13 @@ def _(rid, params: dict) -> dict:
     cfg = _load_cfg()
     model = _resolve_model()
     from agent.secret_scope import get_secret
-
     api_key = get_secret("HERMES_API_KEY", "") or cfg.get("api_key", "")
     masked = f"****{api_key[-4:]}" if len(api_key) > 4 else "(not set)"
     base_url = os.environ.get("HERMES_BASE_URL", "") or cfg.get("base_url", "")
     agent_rows = [
         ["Max Turns", str(_cfg_max_turns(cfg, 500))],
         ["Toolsets", ", ".join(cfg.get("enabled_toolsets", [])) or "all"],
-        ["Verbose", str(cfg.get("verbose", False))],
-    ]
+        ["Verbose", str(cfg.get("verbose", False))]]
     sections = [
         {"title": "Model", "rows": [["Model", model], ["Base URL", base_url or "(default)"], ["API Key", masked]]},
         {"title": "Agent", "rows": agent_rows},
@@ -1205,7 +1099,6 @@ def _(rid, params: dict) -> dict:
 @_guarded(5034)
 def _(rid, params: dict) -> dict:
     from model_tools import get_toolset_for_tool, get_tool_definitions
-
     session = _sessions.get(params.get("session_id", ""))
     enabled = getattr(session["agent"], "enabled_toolsets", None) if session else _load_enabled_toolsets()
     # Pre-assembly list: /tools must also show tools deferred behind the tool_search bridge (as the CLI).
@@ -1222,6 +1115,7 @@ def _(rid, params: dict) -> dict:
 
 
 @method("tools.configure")
+@_guarded(5035)
 def _(rid, params: dict) -> dict:
     action = str(params.get("action", "") or "").strip().lower()
     targets = [str(name).strip() for name in params.get("names", []) or [] if str(name).strip()]
@@ -1229,64 +1123,45 @@ def _(rid, params: dict) -> dict:
         return _err(rid, 4017, f"unknown tools action: {action}")
     if not targets:
         return _err(rid, 4018, "names required")
-
-    try:
-        from hermes_cli.config import load_config, save_config
-        from hermes_cli.tools_config import (
-            CONFIGURABLE_TOOLSETS,
-            _apply_mcp_change,
-            _apply_toolset_change,
-            _get_platform_tools,
-            _get_plugin_toolset_keys,
-        )
-
-        cfg = load_config()
-        valid_toolsets = {ts_key for ts_key, _, _ in CONFIGURABLE_TOOLSETS} | _get_plugin_toolset_keys()
-        toolset_targets = [name for name in targets if ":" not in name]
-        mcp_targets = [name for name in targets if ":" in name]
-        unknown = [name for name in toolset_targets if name not in valid_toolsets]
-        toolset_targets = [name for name in toolset_targets if name in valid_toolsets]
-
-        if toolset_targets:
-            _apply_toolset_change(cfg, "cli", toolset_targets, action)
-
-        missing_servers = _apply_mcp_change(cfg, mcp_targets, action) if mcp_targets else set()
-        save_config(cfg)
-
-        sid = params.get("session_id", "")
-        session = _sessions.get(sid)
-        info = _reset_session_agent(sid, session) if session else None
-        enabled = sorted(_get_platform_tools(load_config(), "cli", include_default_mcp_servers=False))
-        changed = [
-            name
-            for name in targets
-            if name not in unknown and (":" not in name or name.split(":", 1)[0] not in missing_servers)
-        ]
-
-        return _ok(
-            rid,
-            {
-                "changed": changed,
-                "enabled_toolsets": enabled,
-                "info": info,
-                "missing_servers": sorted(missing_servers),
-                "reset": bool(session),
-                "unknown": unknown,
-            },
-        )
-    except Exception as e:
-        return _err(rid, 5035, str(e))
+    from hermes_cli.config import load_config, save_config
+    from hermes_cli.tools_config import (
+        CONFIGURABLE_TOOLSETS, _apply_mcp_change, _apply_toolset_change, _get_platform_tools,
+        _get_plugin_toolset_keys)
+    cfg = load_config()
+    valid_toolsets = {ts_key for ts_key, _, _ in CONFIGURABLE_TOOLSETS} | _get_plugin_toolset_keys()
+    toolset_targets = [name for name in targets if ":" not in name]
+    mcp_targets = [name for name in targets if ":" in name]
+    unknown = [name for name in toolset_targets if name not in valid_toolsets]
+    toolset_targets = [name for name in toolset_targets if name in valid_toolsets]
+    if toolset_targets:
+        _apply_toolset_change(cfg, "cli", toolset_targets, action)
+    missing_servers = _apply_mcp_change(cfg, mcp_targets, action) if mcp_targets else set()
+    save_config(cfg)
+    sid = params.get("session_id", "")
+    session = _sessions.get(sid)
+    info = _reset_session_agent(sid, session) if session else None
+    enabled = sorted(_get_platform_tools(load_config(), "cli", include_default_mcp_servers=False))
+    changed = [
+        name
+        for name in targets
+        if name not in unknown and (":" not in name or name.split(":", 1)[0] not in missing_servers)
+    ]
+    return _ok(rid, {
+        "changed": changed,
+        "enabled_toolsets": enabled,
+        "info": info,
+        "missing_servers": sorted(missing_servers),
+        "reset": bool(session),
+        "unknown": unknown})
 
 
 @method("agents.list")
 @_guarded(5033)
 def _(rid, params: dict) -> dict:
     from tools.process_registry import process_registry
-
     rows = [
         {"session_id": p["session_id"], "command": p["command"][:80], "status": p["status"], "uptime": p["uptime_seconds"]}
-        for p in process_registry.list_sessions()
-    ]
+        for p in process_registry.list_sessions()]
     return _ok(rid, {"processes": rows})
 
 
@@ -1299,7 +1174,6 @@ def _(rid, params: dict) -> dict:
     """cronjob() keys off HERMES_HOME, so the optional ``profile`` scope reaches a
     per-profile cron store even when that profile runs its own gateway."""
     from tools.cronjob_tools import cronjob
-
     action, jid = params.get("action", "list"), params.get("name", "")
     if action == "list":
         # Paused jobs are excluded by default (reads as deletion in a toggle UI) — forward the flag.
@@ -1321,8 +1195,7 @@ def _(rid, params: dict) -> dict:
             prompt=params.get("prompt", ""),
             repeat=int(params["repeat"]) if str(params.get("repeat", "")).strip().isdigit() else None,
             continuity=is_truthy_value(params.get("continuity")) if params.get("continuity") is not None else None,
-            deliver=str(params.get("deliver") or "").strip() or None,
-        )
+            deliver=str(params.get("deliver") or "").strip() or None)
         return _ok(rid, json.loads(raw))
     if action in {"remove", "pause", "resume"}:
         return _ok(rid, json.loads(cronjob(action=action, job_id=jid)))
@@ -1342,46 +1215,34 @@ def _(rid, params: dict) -> dict:
         cols, rows, frames = 80, 24, 48
     from agent.learning_graph import build_learning_graph
     from agent.learning_graph_render import render_frames
-
     return _ok(rid, render_frames(build_learning_graph(), cols=max(20, cols), rows=max(10, rows), frames=frames))
 
 
-@method("learning.detail")
-@_guarded(5000, "learning.detail failed: ")
-def _(rid, params: dict) -> dict:
-    """Current content of a journey node, for an edit prefill."""
-    from agent.learning_mutations import node_detail
+def _learning_mutation(fn_name: str, arg_keys: tuple):
+    """learning.* body: ``agent.learning_mutations.<fn_name>(*str(params[k]) for k in arg_keys)``."""
 
-    return _ok(rid, node_detail(str(params.get("id", ""))))
+    def body(rid, params: dict) -> dict:
+        import agent.learning_mutations as mutations
+        return _ok(rid, getattr(mutations, fn_name)(*(str(params.get(k, "")) for k in arg_keys)))
+    return body
 
 
-@method("learning.delete")
-@_guarded(5000, "learning.delete failed: ")
-def _(rid, params: dict) -> dict:
-    """Delete a journey node — skills are archived (restorable), memories removed."""
-    from agent.learning_mutations import delete_node
-
-    return _ok(rid, delete_node(str(params.get("id", ""))))
-
-
-@method("learning.edit")
-@_guarded(5000, "learning.edit failed: ")
-def _(rid, params: dict) -> dict:
-    """Rewrite a journey node's content (SKILL.md or memory chunk)."""
-    from agent.learning_mutations import edit_node
-
-    return _ok(rid, edit_node(str(params.get("id", "")), str(params.get("content", ""))))
+# detail → node content for an edit prefill; delete → skills archived (restorable), memories
+# removed; edit → rewrite a node's content (SKILL.md or memory chunk).
+for _rpc, _fn, _keys in (
+    ("detail", "node_detail", ("id",)), ("delete", "delete_node", ("id",)), ("edit", "edit_node", ("id", "content")),
+):
+    method(f"learning.{_rpc}")(_guarded(5000, f"learning.{_rpc} failed: ")(_learning_mutation(_fn, _keys)))
+del _rpc, _fn, _keys
 
 
 def _skills_list(rid, params, query):
     from hermes_cli.banner import get_available_skills
-
     return _ok(rid, {"skills": get_available_skills()})
 
 
 def _skills_search(rid, params, query):
     from tools.skills_hub import GitHubAuth, create_source_router, unified_search
-
     raw = unified_search(query, create_source_router(GitHubAuth()), source_filter="all", limit=20) or []
     return _ok(rid, {"results": [{"name": r.name, "description": r.description} for r in raw]})
 
@@ -1392,21 +1253,18 @@ def _skills_install(rid, params, query):
     class _Q:
         def print(self, *a, **k):
             pass
-
     do_install(query, skip_confirm=True, console=_Q())
     return _ok(rid, {"installed": True, "name": query})
 
 
 def _skills_browse(rid, params, query):
     from hermes_cli.skills_hub import browse_skills
-
     pg = int(params.get("page", 0) or 0) or (int(query) if query.isdigit() else 1)
     return _ok(rid, browse_skills(page=pg, page_size=int(params.get("page_size", 20))))
 
 
 def _skills_inspect(rid, params, query):
     from hermes_cli.skills_hub import inspect_skill
-
     return _ok(rid, {"info": inspect_skill(query) or {}})
 
 
@@ -1431,7 +1289,6 @@ def _(rid, params: dict) -> dict:
 @_guarded(5025)
 def _(rid, params: dict) -> dict:
     from agent.skill_commands import reload_skills
-
     result = reload_skills()
     added = result.get("added") or []
     removed = result.get("removed") or []
@@ -1457,7 +1314,6 @@ def _(rid, params: dict) -> dict:
     """``{servers: [{name, description, installed, enabled, requires: [env keys], transport}]}``
     — the `hermes mcp` menu with per-profile state, so UIs know which entries need setup."""
     from hermes_cli import mcp_catalog
-
     out = []
     for entry in mcp_catalog.list_catalog():
         try:
@@ -1472,9 +1328,7 @@ def _(rid, params: dict) -> dict:
                 "installed": bool(mcp_catalog.is_installed(entry.name)),
                 "enabled": bool(mcp_catalog.is_enabled(entry.name)),
                 "requires": requires,
-                "transport": str(getattr(transport, "kind", "") or transport or "stdio"),
-            }
-        )
+                "transport": str(getattr(transport, "kind", "") or transport or "stdio")})
     return _ok(rid, {"servers": out})
 
 
@@ -1484,7 +1338,6 @@ def _(rid, params: dict) -> dict:
     """``{servers: [{name, transport, url, command, args, env (key names only),
     auth, oauth_tokens_present, enabled, tools}]}`` for the scoped profile."""
     from hermes_cli.mcp_config import _get_mcp_servers
-
     servers = _get_mcp_servers()
     return _ok(rid, {"servers": [_mcp_summarize_server(name, cfg) for name, cfg in sorted(servers.items())]})
 
@@ -1496,7 +1349,6 @@ def _(rid, params: dict) -> dict:
     headers/auth/tools). ``bearer_token`` goes to the profile's .env; only the
     ``Authorization`` header template is persisted. Duplicate names → 4090."""
     from hermes_cli.mcp_config import _apply_mcp_preset, _get_mcp_servers, _save_bearer_auth_token, _save_mcp_server
-
     name = str(params.get("name") or "").strip()
     if name in _get_mcp_servers():
         return _err(rid, 4090, f"server '{name}' already exists")
@@ -1510,9 +1362,7 @@ def _(rid, params: dict) -> dict:
             url=server_config.get("url"),
             command=server_config.get("command"),
             cmd_args=list(server_config.get("args") or []),
-            server_config=server_config,
-        )
-
+            server_config=server_config)
     if not server_config.get("url") and not server_config.get("command"):
         return _err(rid, 4063, "config must specify a 'url' (http) or 'command' (stdio), or a valid 'preset'")
     bearer_token = params.get("bearer_token")
@@ -1532,7 +1382,6 @@ def _(rid, params: dict) -> dict:
     (stdio), matching ``cmd_mcp_configure`` / ``_save_bearer_auth_token``."""
     from hermes_cli.config import load_config, save_config, save_env_value
     from hermes_cli.mcp_config import _bearer_auth_headers, _env_key_for_server, _strip_bearer_prefix
-
     name, servers, err = _mcp_named_server(rid, params)
     if err:
         return err
@@ -1541,7 +1390,6 @@ def _(rid, params: dict) -> dict:
     entry = servers[name]
     if not isinstance(entry, dict):
         return _err(rid, 4001, "malformed server config")
-
     if entry.get("url"):
         normalized = _strip_bearer_prefix(str(value))
         if not normalized or normalized.lower() == "bearer":
@@ -1558,7 +1406,6 @@ def _(rid, params: dict) -> dict:
             env_block = {}
         env_block[env_var] = f"${{{env_var}}}"
         entry["env"] = env_block
-
     cfg = load_config()
     cfg.setdefault("mcp_servers", {})[name] = entry
     save_config(cfg)
@@ -1572,7 +1419,6 @@ def _(rid, params: dict) -> dict:
     oauth_tokens_present}``; failure: ``{ok: false, error, tools: [], oauth_needed, ...}``.
     Runs on the RPC pool (_LONG_HANDLERS): a cold stdio `npx` spawn can block for seconds."""
     from hermes_cli.mcp_config import _oauth_tokens_present, _probe_single_server
-
     name, servers, err = _mcp_named_server(rid, params)
     if err:
         return err
@@ -1585,7 +1431,6 @@ def _(rid, params: dict) -> dict:
     def failure(error: str, oauth_needed: bool, tokens_present) -> dict:
         payload = {"ok": False, "error": error, "tools": [], "oauth_needed": oauth_needed}
         return _ok(rid, {**payload, "oauth_tokens_present": tokens_present})
-
     try:
         tools = _probe_single_server(name, cfg, details=details)
         token_present = _oauth_tokens_present(name) if needs_oauth_token else True
@@ -1599,8 +1444,7 @@ def _(rid, params: dict) -> dict:
         "prompts": details.get("prompts", 0),
         "resources": details.get("resources", 0),
         "oauth_needed": needs_oauth_token,
-        "oauth_tokens_present": True if needs_oauth_token else None,
-    }
+        "oauth_tokens_present": True if needs_oauth_token else None}
     return _ok(rid, payload)
 
 
@@ -1609,7 +1453,6 @@ def _(rid, params: dict) -> dict:
 def _(rid, params: dict) -> dict:
     """Remove a server from the profile's config.yaml → ``{ok: true, removed: true}``."""
     from hermes_cli.mcp_config import _remove_mcp_server
-
     name = str(params.get("name") or "").strip()
     if not _remove_mcp_server(name):
         return _err(rid, 4064, f"server '{name}' not found")
@@ -1630,7 +1473,6 @@ def _(rid, params: dict) -> dict:
     try:
         from hermes_constants import get_hermes_home
         from tui_gateway import mcp_oauth_sessions
-
         name, servers, err = _mcp_named_server(rid, params)
         if err:
             return err
@@ -1640,7 +1482,6 @@ def _(rid, params: dict) -> dict:
         if cfg.get("headers") and cfg.get("auth") != "oauth":
             return _err(rid, 4001, "this server uses header/API-key auth, not OAuth")
         cfg["auth"] = "oauth"
-
         hermes_home = str(get_hermes_home().expanduser().resolve(strict=False))
         result = mcp_oauth_sessions.start_flow(hermes_home, name, cfg, client_redirect_uri=client_redirect_uri)
     except ValueError as e:
@@ -1654,7 +1495,6 @@ def _(rid, params: dict) -> dict:
     """Poll a flow → ``{ok, status: pending|approved|error, error_message?, auth_url?, tools?}``.
     On ``approved`` tokens persist for that server/profile (profile scope applies here too)."""
     from tui_gateway import mcp_oauth_sessions
-
     name = str(params.get("name") or "").strip()
     session_id = str(params.get("session_id") or "").strip()
     result = mcp_oauth_sessions.poll_flow(session_id, name)
@@ -1667,7 +1507,6 @@ def _(rid, params: dict) -> dict:
     """Relay a client-captured redirect (``code``/``state``/``error``) into a flow started with
     ``client_redirect_uri``. ``{ok: true}`` once accepted (state verified), else ``{ok: false, error_message}``."""
     from tui_gateway import mcp_oauth_sessions
-
     name = str(params.get("name") or "").strip()
     session_id = str(params.get("session_id") or "").strip()
     result = mcp_oauth_sessions.deliver_callback_flow(
@@ -1675,8 +1514,7 @@ def _(rid, params: dict) -> dict:
         name,
         code=str(params.get("code") or "") or None,
         state=str(params.get("state") or "") or None,
-        error=str(params.get("error") or "") or None,
-    )
+        error=str(params.get("error") or "") or None)
     return _ok(rid, result)
 
 
@@ -1690,9 +1528,7 @@ def _plugin_rows() -> list[dict]:
         _get_disabled_set,
         _get_enabled_set,
         _is_portable_plugin_dir,
-        _plugin_status,
-    )
-
+        _plugin_status)
     enabled = _get_enabled_set()
     disabled = _get_disabled_set()
     out = []
@@ -1711,8 +1547,7 @@ def _plugin_rows() -> list[dict]:
                 "source": source,
                 "status": status,
                 "portable": _is_portable_plugin_dir(_dir),  # Agent Plugins v1 package vs native Hermes plugin
-            }
-        )
+            })
     return out
 
 
@@ -1738,7 +1573,6 @@ def _plugins_toggle(rid, params):
 
 def _plugins_install(rid, params):
     from hermes_cli.plugins_cmd import dashboard_install_plugin
-
     ident = (params.get("identifier") or params.get("repo") or "").strip()
     if not ident:
         return _err(rid, 4019, "plugins.install requires 'identifier' or 'repo'")
@@ -1770,7 +1604,6 @@ def _(rid, params: dict) -> dict:
         return _err(rid, 4004, "empty command")
     try:
         from tools.approval import detect_dangerous_command, detect_hardline_command
-
         is_hardline, hardline_desc = detect_hardline_command(cmd)
         if is_hardline:
             return _err(rid, 4005, f"blocked (hardline): {hardline_desc}. Use the agent for dangerous commands.")