refactor(tui_gateway): projects mutator table, mcp_startup call helper, kwargs table in agent_callbacks

- methods_projects: update/add_folder/remove_folder/set_primary handlers registered from a
  _PROJECT_MUTATORS table (same @method names, same 5062/5063/5061 mapping); _pick() for
  params->kwargs; old-vs-new RPC golden (27 calls incl. error paths) identical.
- entry: _mcp_startup_call() replaces 5 lazy import+try/except ladders on hermes_cli.mcp_startup
  (tests monkeypatch mcp_startup module attrs — still resolved at call time).
- agent_callbacks: _background_agent_kwargs built from key tables; drop unused _registry.
- docstring compaction (module docstrings <= 8 lines, WHYs kept).
WIRE-PARITY-OK.

tui_gateway/agent_callbacks.py

@@ -9,9 +9,7 @@ from __future__ import annotations
 
 import threading
 
-from .method_ctx import HandlerRegistry, bind_module
-
-_registry = HandlerRegistry()
+from .method_ctx import bind_module
 
 
 # Child-session live mirror: a delegated child's activity reaches the gateway only
@@ -204,11 +202,9 @@ def _available_personalities(cfg: dict | None = None) -> dict:
 
 
 def _validate_personality(value: str, cfg: dict | None = None) -> tuple[str, str]:
-    """Resolve a requested personality to (name, prompt) or raise ValueError.
-
-    Same contract as hermes_cli.personality.resolve_personality, but goes through
-    the module-level _available_personalities so tests keep a single patch point.
-    """
+    """Resolve a requested personality to (name, prompt) or raise ValueError. Same
+    contract as hermes_cli.personality.resolve_personality, but goes through the
+    module-level _available_personalities so tests keep a single patch point."""
     from hermes_cli.personality import normalize_personality_name, render_personality_prompt
 
     name = normalize_personality_name(value)
@@ -230,12 +226,10 @@ def _prompt_text(value) -> str:
 
 def _apply_personality_to_session(
     sid: str, session: dict, new_prompt: str, personality: str = "") -> tuple[bool, dict | None]:
-    """Apply a personality change to a live session without resetting history.
-
-    Updates the ephemeral system prompt in place (appended at API-call time, so
-    prompt-cache hits survive) and injects a pivot marker so the model stops
-    pattern-matching its earlier tone. Returns (history_reset=False, info).
-    """
+    """Apply a personality change to a live session without resetting history: the
+    ephemeral system prompt is updated in place (appended at API-call time, so
+    prompt-cache hits survive) plus a pivot marker so the model stops pattern-matching
+    its earlier tone. Returns (history_reset=False, info)."""
     if not session:
         return False, None
     session["personality"] = personality
@@ -287,9 +281,8 @@ def _parse_tui_skills_env() -> list[str]:
 
 
 def _load_fallback_model():
-    """Configured fallback chain for TUI-created agents, via the shared
-    ``get_fallback_chain`` (parity with HermesCLI/gateway: ``fallback_providers``
-    first in order, legacy ``fallback_model`` merged after, deduped)."""
+    """Configured fallback chain via the shared ``get_fallback_chain`` (parity with
+    HermesCLI/gateway: ``fallback_providers`` first, legacy ``fallback_model`` merged after)."""
     from hermes_cli.fallback_config import get_fallback_chain
 
     return get_fallback_chain(_load_cfg())
@@ -308,36 +301,26 @@ def _background_agent_kwargs(agent, task_id: str) -> dict:
     def g(name, default=None):
         return getattr(agent, name, default)
 
-    return {
-        "base_url": g("base_url") or None,
-        "api_key": g("api_key") or None,
-        "provider": g("provider") or None,
-        "api_mode": g("api_mode") or None,
-        "acp_command": g("acp_command") or None,
-        "acp_args": g("acp_args") or None,
-        "model": g("model") or _resolve_model(),
-        "max_iterations": _cfg_max_turns(cfg, 25),
+    kwargs = {k: g(k) or None for k in (
+        "base_url", "api_key", "provider", "api_mode", "acp_command", "acp_args", "ephemeral_system_prompt")}
+    kwargs.update({k: g(k) for k in (
+        "providers_allowed", "providers_ignored", "providers_order", "provider_sort",
+        "provider_data_collection", "openrouter_min_coding_score")})
+    kwargs.update(
+        model=g("model") or _resolve_model(),
+        max_iterations=_cfg_max_turns(cfg, 25),
         # Detached tasks declare platform="tui" (no UI sid for renderer-routed
         # events), so resolve toolsets against it — never GUI schema they can't use.
-        "enabled_toolsets": g("enabled_toolsets") or _load_enabled_toolsets("tui"),
-        "quiet_mode": True,
-        "verbose_logging": False,
-        "ephemeral_system_prompt": g("ephemeral_system_prompt") or None,
-        "providers_allowed": g("providers_allowed"),
-        "providers_ignored": g("providers_ignored"),
-        "providers_order": g("providers_order"),
-        "provider_sort": g("provider_sort"),
-        "provider_require_parameters": g("provider_require_parameters", False),
-        "provider_data_collection": g("provider_data_collection"),
-        "openrouter_min_coding_score": g("openrouter_min_coding_score"),
-        "session_id": task_id,
-        "reasoning_config": g("reasoning_config")
-        or _load_reasoning_config(str(g("model", "") or "")),
-        "service_tier": g("service_tier") or _load_service_tier(),
-        "request_overrides": dict(g("request_overrides", {}) or {}),
-        "platform": "tui",
-        "session_db": _get_db(),
-        "fallback_model": _agent_fallback_model(agent)}
+        enabled_toolsets=g("enabled_toolsets") or _load_enabled_toolsets("tui"),
+        quiet_mode=True, verbose_logging=False,
+        provider_require_parameters=g("provider_require_parameters", False),
+        session_id=task_id,
+        reasoning_config=g("reasoning_config") or _load_reasoning_config(str(g("model", "") or "")),
+        service_tier=g("service_tier") or _load_service_tier(),
+        request_overrides=dict(g("request_overrides", {}) or {}),
+        platform="tui", session_db=_get_db(), fallback_model=_agent_fallback_model(agent),
+    )
+    return kwargs
 
 
 def _ephemeral_preview_agent_kwargs(agent, task_id: str) -> dict:
@@ -347,10 +330,9 @@ def _ephemeral_preview_agent_kwargs(agent, task_id: str) -> dict:
 
 
 def _preview_restart_history(session: dict, max_messages: int = 24, max_tool_chars: int = 1200) -> list[dict]:
-    """Distill recent parent history for the ephemeral preview-restart agent, which
-    otherwise would guess app/server/cwd/port from the bare URL + console logs.
-    Keeps the last ``max_messages`` (always back to the last user turn); tool
-    results are truncated so file dumps don't blow the context window."""
+    """Distill recent parent history for the ephemeral preview-restart agent (else it
+    guesses app/server/cwd/port from the bare URL). Keeps the last ``max_messages``
+    (always back to the last user turn); tool results truncated to ``max_tool_chars``."""
     try:
         with session["history_lock"]:
             history = list(session.get("history", []) or [])

tui_gateway/entry.py

@@ -64,6 +64,16 @@ def _stamp() -> str:
     return time.strftime("%Y-%m-%d %H:%M:%S")
 
 
+def _mcp_startup_call(name: str, *args, default=None, **kwargs):
+    """Call ``hermes_cli.mcp_startup.<name>`` (lazy import); ``default`` on any failure."""
+    try:
+        from hermes_cli import mcp_startup
+
+        return getattr(mcp_startup, name)(*args, **kwargs)
+    except Exception:
+        return default
+
+
 def _append_crash_log(header: str, dump=None) -> None:
     """Best-effort ``=== header ===`` entry in the crash log; ``dump(f)`` adds detail."""
     with suppress(Exception):
@@ -165,13 +175,8 @@ def wait_for_mcp_discovery(timeout: "float | None" = None) -> None:
     """
     thread = _mcp_discovery_thread
     if thread is not None and thread.is_alive():
-        try:
-            from hermes_cli.mcp_startup import _resolve_discovery_timeout
-
-            bound = _resolve_discovery_timeout(timeout)
-        except Exception:
-            bound = timeout if timeout is not None else 0.75
-        thread.join(timeout=bound)
+        fallback = timeout if timeout is not None else 0.75
+        thread.join(timeout=_mcp_startup_call("_resolve_discovery_timeout", timeout, default=fallback))
         return
     # Shared-owner path. Re-invoke the idempotent spawn first so a previous
     # zero-connected run gets its retry instead of latching the process MCP-less.
@@ -186,10 +191,7 @@ def wait_for_mcp_discovery(timeout: "float | None" = None) -> None:
         start_background_mcp_discovery(logger=logger, thread_name="tui-mcp-discovery")
     except Exception:
         logger.debug("TUI MCP discovery retry-spawn failed", exc_info=True)
-    with suppress(Exception):
-        from hermes_cli.mcp_startup import wait_for_mcp_discovery as _startup_wait
-
-        _startup_wait(timeout)
+    _mcp_startup_call("wait_for_mcp_discovery", timeout)
 
 
 def mcp_discovery_in_flight() -> bool:
@@ -200,12 +202,7 @@ def mcp_discovery_in_flight() -> bool:
     thread = _mcp_discovery_thread
     if thread is not None and thread.is_alive():
         return True
-    try:
-        from hermes_cli.mcp_startup import mcp_discovery_in_flight as _startup_in_flight
-
-        return _startup_in_flight()
-    except Exception:
-        return False
+    return _mcp_startup_call("mcp_discovery_in_flight", default=False)
 
 
 def join_mcp_discovery(timeout: float | None = None) -> bool:
@@ -217,13 +214,7 @@ def join_mcp_discovery(timeout: float | None = None) -> bool:
     if thread is not None:
         thread.join(timeout=timeout)
         entry_done = not thread.is_alive()
-    try:
-        from hermes_cli.mcp_startup import join_mcp_discovery as _startup_join
-
-        startup_done = _startup_join(timeout=timeout)
-    except Exception:
-        startup_done = True
-    return entry_done and startup_done
+    return entry_done and _mcp_startup_call("join_mcp_discovery", timeout=timeout, default=True)
 
 
 # Spurious stdin-EOF recovery tracker (shared open-file-description O_NONBLOCK flip).
@@ -242,11 +233,10 @@ def ensure_mcp_discovery_started() -> None:
 
     ``main()`` calls this for stdio; WS/Desktop skip ``main()``, so
     ``server._start_agent_build`` also calls it AFTER binding the session profile's
-    HERMES_HOME — the shared owner captures that context-local override, so
-    discovery reads the SELECTED profile's ``mcp_servers``. Delegating keeps the
-    process-wide start lock, retry-after-zero-connected allowance and
-    interactive-OAuth suppression. MCP registration is process-global: the FIRST
-    profile to build an agent wins the discovery slot.
+    HERMES_HOME (the shared owner captures that override, so discovery reads the
+    SELECTED profile's ``mcp_servers``). Delegating keeps the process-wide start lock,
+    retry-after-zero-connected allowance and interactive-OAuth suppression. MCP
+    registration is process-global: the FIRST profile to build an agent wins.
     """
     global _mcp_discovery_enabled
 

tui_gateway/mcp_oauth_sessions.py

@@ -1,14 +1,11 @@
 """Session-backed MCP OAuth flows for the gateway (mcp.servers.oauth.*).
 
-Mirrors the dashboard's *provider* OAuth model: ``start`` kicks off a background
-worker and returns ``{session_id, auth_url, flow}``; ``poll`` reports
-``{status: pending|approved|error}`` until tokens land on disk. No OAuth logic is
-reimplemented — the token machinery is ``hermes mcp login``'s
-(``_probe_single_server`` under ``force_interactive_oauth``) and ``DashboardOAuthFlow``
-is the thread-safe bridge; the only new piece is a loopback HTTP listener feeding
-``deliver_callback``. Remote-backend variant: the client binds its OWN loopback
-listener, passes ``client_redirect_uri`` to ``start`` and relays the redirect via
-``deliver_callback_flow``; state verification stays server-side either way.
+``start`` kicks off a background worker and returns ``{session_id, auth_url, flow}``;
+``poll`` reports ``{status: pending|approved|error}`` until tokens land on disk. No OAuth
+logic is reimplemented: ``hermes mcp login``'s probe under ``force_interactive_oauth``
+plus ``DashboardOAuthFlow`` as the bridge; the only new piece is a loopback listener
+feeding ``deliver_callback``. Remote backends: the client hosts the listener, passes
+``client_redirect_uri`` and relays via ``deliver_callback_flow`` (state check stays here).
 """
 
 from __future__ import annotations

tui_gateway/methods_projects.py

@@ -63,12 +63,35 @@ def _require_project(pdb, conn, params: dict):
     return proj
 
 
-def _project_ok(rid, pdb, conn, pid) -> dict:
-    return _ok(rid, {"project": pdb.get_project(conn, pid).to_dict()})
+def _pick(params: dict, *keys: str) -> dict:
+    return {k: params.get(k) for k in keys}
 
 
-def _path_param(params: dict) -> str:
-    return str(params.get("path") or "")
+# Per-project mutators: (rpc suffix, pdb function, takes params['path'], extra kwargs).
+# Each resolves ``params['id']`` (5062 when missing), mutates, and answers with the
+# refreshed project.
+_PROJECT_MUTATORS = (
+    ("update", "update_project", False,
+     lambda p: _pick(p, "name", "description", "icon", "color", "board_slug")),
+    ("add_folder", "add_folder", True,
+     lambda p: {"label": p.get("label"), "is_primary": bool(p.get("is_primary"))}),
+    ("remove_folder", "remove_folder", True, lambda p: {}),
+    ("set_primary", "set_primary", True, lambda p: {}),
+)
+
+
+def _register_project_mutator(suffix: str, fn_name: str, takes_path: bool, kwargs_of) -> None:
+    @_projects_method(f"projects.{suffix}")
+    def _(rid, params, pdb, conn) -> dict:
+        proj = _require_project(pdb, conn, params)
+        args = (str(params.get("path") or ""),) if takes_path else ()
+        getattr(pdb, fn_name)(conn, proj.id, *args, **kwargs_of(params))
+        return _ok(rid, {"project": pdb.get_project(conn, proj.id).to_dict()})
+
+
+for _spec in _PROJECT_MUTATORS:
+    _register_project_mutator(*_spec)
+del _spec
 
 
 @_projects_method("projects.list")
@@ -84,49 +107,14 @@ def _(rid, params, pdb, conn) -> dict:
 @_projects_method("projects.create")
 def _(rid, params, pdb, conn) -> dict:
     pid = pdb.create_project(
-        conn, name=str(params.get("name") or ""), slug=params.get("slug"),
-        folders=params.get("folders") or [], primary_path=params.get("primary_path"),
-        description=params.get("description"), icon=params.get("icon"), color=params.get("color"),
-        board_slug=params.get("board_slug"))
+        conn, name=str(params.get("name") or ""), folders=params.get("folders") or [],
+        **_pick(params, "slug", "primary_path", "description", "icon", "color", "board_slug"))
     if params.get("use"):
         pdb.set_active(conn, pid)
     proj = pdb.get_project(conn, pid)
     return _ok(rid, {"project": proj.to_dict() if proj else None})
 
 
-@_projects_method("projects.update")
-def _(rid, params, pdb, conn) -> dict:
-    proj = _require_project(pdb, conn, params)
-    pdb.update_project(
-        conn, proj.id, name=params.get("name"), description=params.get("description"),
-        icon=params.get("icon"), color=params.get("color"), board_slug=params.get("board_slug"))
-    return _project_ok(rid, pdb, conn, proj.id)
-
-
-@_projects_method("projects.add_folder")
-def _(rid, params, pdb, conn) -> dict:
-    proj = _require_project(pdb, conn, params)
-    pdb.add_folder(
-        conn, proj.id, _path_param(params), label=params.get("label"),
-        is_primary=bool(params.get("is_primary")),
-    )
-    return _project_ok(rid, pdb, conn, proj.id)
-
-
-@_projects_method("projects.remove_folder")
-def _(rid, params, pdb, conn) -> dict:
-    proj = _require_project(pdb, conn, params)
-    pdb.remove_folder(conn, proj.id, _path_param(params))
-    return _project_ok(rid, pdb, conn, proj.id)
-
-
-@_projects_method("projects.set_primary")
-def _(rid, params, pdb, conn) -> dict:
-    proj = _require_project(pdb, conn, params)
-    pdb.set_primary(conn, proj.id, _path_param(params))
-    return _project_ok(rid, pdb, conn, proj.id)
-
-
 @_projects_method("projects.archive")
 def _(rid, params, pdb, conn) -> dict:
     proj = _require_project(pdb, conn, params)
@@ -136,8 +124,7 @@ def _(rid, params, pdb, conn) -> dict:
 
 @_projects_method("projects.delete")
 def _(rid, params, pdb, conn) -> dict:
-    proj = _require_project(pdb, conn, params)
-    pdb.delete_project(conn, proj.id)
+    pdb.delete_project(conn, _require_project(pdb, conn, params).id)
     return _ok(rid, _projects_payload(conn))
 
 
@@ -155,14 +142,10 @@ def _(rid, params, pdb, conn) -> dict:
 
 
 def _non_workspace_dirs() -> set[str]:
-    """Directories that are never a workspace: ``/``, the user's home, and the dir
-    homes live in (``/home``, ``/Users``, ``C:\\Users``).
-
-    Both POSIX spellings are excluded on every host because both are reachable as
-    a cwd anywhere (macOS ships an empty ``/home`` autofs stub; containers/remote
-    shells hand back Linux paths). Promoting one mints a catch-all project, and
-    ``/home`` renders as a second "home" row next to the Home bucket.
-    """
+    """Never-a-workspace dirs: ``/``, the user's home, and the dir homes live in. Both
+    POSIX spellings are excluded on every host (macOS ships an empty ``/home`` autofs
+    stub; containers/remote shells hand back Linux paths) — promoting one mints a
+    catch-all project and ``/home`` renders as a second "home" row beside Home."""
     home = os.path.realpath(os.path.expanduser("~"))
     candidates = (os.sep, home, os.path.dirname(home), "/home", "/Users")
     return {os.path.normcase(os.path.realpath(path)) for path in candidates if path}
@@ -246,18 +229,11 @@ def _repo_discovery_policy_is_default(policy: dict) -> bool:
 
 
 def _scan_discovered_repos_remote(conn, policy: dict) -> bool:
-    """Backend-side disk scan of the discovery policy roots into the discovery cache.
-
-    The desktop's native scan only sees the local filesystem; on a remote gateway
-    the host must scan its own disk so zero-session repos still appear. Walk each
-    root (bounded), record ``.git``-bearing dirs. Best-effort: failures log and
-    leave the cache untouched.
-
-    Returns True only when the scan is authoritative (every root walked to
-    completion, cap not hit) — only then is the cache write ``replace=True``. A
-    partial/errored scan must MERGE, never wipe, so a failed remote refresh can't
-    blank the sidebar.
-    """
+    """Backend-side disk scan of the policy roots into the discovery cache (the desktop's
+    native scan only sees the local filesystem). Best-effort: failures log and leave
+    the cache untouched. Returns True only when the scan is authoritative (every root
+    walked to completion, cap not hit) — only then is the cache write ``replace=True``;
+    a partial/errored scan must MERGE, never wipe, or a failed refresh blanks the sidebar."""
     from hermes_cli import projects_db as pdb
     roots = policy.get("roots") or []
     excludes = policy.get("exclude_paths") or []
@@ -310,14 +286,10 @@ def _scan_discovered_repos_remote(conn, policy: dict) -> bool:
 
 def _discover_repos_payload(
     db, *, conn=None, backfill: bool = True, include_cached: bool = True) -> list[dict]:
-    """Merge filesystem-scanned repos (cached) with session-derived repo roots.
-
-    Repo-first: the disk scan surfaces repos with zero sessions; session-derived
-    roots cover repos outside the scan roots. Both are junk-filtered and carry
-    session totals. ``conn`` reuses an open projects.db connection; ``backfill``
-    persists resolved roots onto session rows — kept OFF the per-turn tree path
-    (grouping uses the live git resolver) and done only on explicit refresh.
-    """
+    """Merge filesystem-scanned repos (cached; may have zero sessions) with
+    session-derived roots, junk-filtered, with session totals. ``conn`` reuses an open
+    projects.db connection; ``backfill`` persists resolved roots onto session rows —
+    kept OFF the per-turn tree path and done only on explicit refresh."""
     repos: dict[str, dict] = {}
 
     def _agg(root: str) -> dict:
@@ -393,11 +365,8 @@ def _project_tree_inputs(
     db, session_limit: int, *, include_discovered: bool
 ) -> tuple[list[dict], list[dict], list[dict], str | None]:
     """Gather (sessions, projects, discovered_repos, active_id) for build_tree.
-
-    ``include_discovered`` is the zero-session-repo overview tier; the drill-in
-    view skips it (its project already has sessions), avoiding the distinct-cwd
-    scan + git probes on that per-turn path.
-    """
+    ``include_discovered`` is the zero-session-repo overview tier; drill-in skips it,
+    avoiding the distinct-cwd scan + git probes on that per-turn path."""
     rows = db.list_sessions_rich(
         limit=session_limit,
         offset=0,

tui_gateway/project_tree.py

@@ -458,14 +458,11 @@ def build_tree(
     exists: Optional[Exists] = None) -> dict:
     """Build the authoritative project tree -> ``{"projects", "scoped_session_ids"}``.
 
-    ``projects`` are ``Project.to_dict()`` shapes; ``sessions`` projected row dicts;
-    ``discovered_repos`` ``{"root", "label", "sessions", "last_active"}``. ``is_junk_root``
-    flags git roots that must never become an AUTO project (bare home, HERMES_HOME);
-    ``is_junk_cwd`` is the narrower policy for non-git folders. User-created projects
-    are honored regardless. ``exists`` keeps a DELETED workspace from becoming a
-    phantom AUTO project; omit it (remote backends) to keep every candidate.
-    ``hydrate`` False (overview) empties lane ``sessions`` but keeps counts and adds
-    up to ``preview_limit`` ``previewSessions``; True (drill-in) keeps full rows.
+    ``is_junk_root`` flags git roots that must never become an AUTO project (bare home,
+    HERMES_HOME); ``is_junk_cwd`` is the narrower policy for non-git folders; explicit
+    projects are honored regardless. ``exists`` keeps a DELETED workspace from becoming
+    a phantom AUTO project (omit on remote backends). ``hydrate`` False (overview) empties
+    lane ``sessions`` but keeps counts + ``preview_limit`` ``previewSessions``.
     """
     active_projects = [p for p in projects if not p.get("archived")]
     _junk = is_junk_root or (lambda _root: False)