From 79d0d3b600fb1f2d224748987130ba47b5a5ef26 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:24:24 -0700
Subject: [PATCH] refactor(tui_gateway): compact
 project_tree/methods_projects/agent_callbacks/entry/mcp_oauth_sessions (-16%)

- project_tree: split build_tree (207 LOC) into _auto_buckets/_home_project phase helpers;
  _field() replaces 9 '(x.get(k) or "").strip()' ladders; _project_node takes wire-shaped
  **flags; drop unused placement 'repo_path' and _strip_trailing_sep; _FolderIndex/
  _project_for_session collapsed. Old-vs-new golden (67 synthetic build_tree runs) identical.
- methods_projects: _project_ok/_path_param unify 4 handler tails; _project_tree_row via
  dict comprehensions; discovered-cache read via contextlib.nullcontext; policy loader
  helpers. Golden over rows/policy/junk predicates identical.
- agent_callbacks: subagent mirror dispatch on a delta table; drop _render_personality_prompt
  (single caller); getattr shorthand in _background_agent_kwargs.
- entry: _write_or_exit unifies 3 write-fail exits; suppress(); heartbeat/sweep start loop.
- mcp_oauth_sessions: suppress(), gc/listener/redirect compaction, docstrings.
WIRE-PARITY-OK; tests green.
---
 tui_gateway/agent_callbacks.py    | 194 ++++++-------
 tui_gateway/entry.py              | 224 ++++++---------
 tui_gateway/mcp_oauth_sessions.py | 215 ++++----------
 tui_gateway/methods_projects.py   | 301 ++++++++------------
 tui_gateway/project_tree.py       | 457 ++++++++++--------------------
 5 files changed, 501 insertions(+), 890 deletions(-)

diff --git a/tui_gateway/agent_callbacks.py b/tui_gateway/agent_callbacks.py
index 7a9d586dfb..6654a2dd79 100644
--- a/tui_gateway/agent_callbacks.py
+++ b/tui_gateway/agent_callbacks.py
@@ -1,4 +1,5 @@
-"""Agent callback wiring: child-session live mirror, per-session agent callbacks, personality overlay, background/preview agent kwargs, agent reset.
+"""Agent callback wiring: child-session live mirror, per-session agent callbacks,
+personality overlay, background/preview agent kwargs, agent reset.
 
 Bodies are rebound onto server.py's globals at install time (see
 method_ctx.bind_module), so they reference server.py globals bare.
@@ -52,37 +53,33 @@ def _mirror_subagent_to_child(event_type: str, payload: dict) -> None:
         return
     csid = live[0]
     text = str(payload.get("text") or "")
+    # thinking/text/start (the child's goal, as a one-time header) are plain deltas.
+    delta = {"subagent.thinking": "reasoning.delta", "subagent.text": "message.delta",
+             "subagent.start": "message.delta"}
     with _child_mirrors_lock:
         st = _child_mirrors.setdefault(child_key, {"seq": 0, "open_tool": None, "started": False})
         if not st["started"]:
             st["started"] = True
             _emit("message.start", csid)
-        if event_type == "subagent.thinking":
+        if event_type in delta:
             if text:
-                _emit("reasoning.delta", csid, {"text": text})
-        elif event_type == "subagent.text":
-            if text:
-                _emit("message.delta", csid, {"text": text})
-        elif event_type == "subagent.start":
-            # One-time header (the child's goal) so a fresh window has context first.
-            if text:
-                _emit("message.delta", csid, {"text": f"{text}\n"})
-        elif event_type == "subagent.tool":
-            if st["open_tool"]:
-                _emit("tool.complete", csid, st["open_tool"])
+                if event_type == "subagent.start":
+                    text = f"{text}\n"
+                _emit(delta[event_type], csid, {"text": text})
+            return
+        if event_type not in ("subagent.tool", "subagent.complete"):
+            return
+        if st["open_tool"]:
+            _emit("tool.complete", csid, st["open_tool"])
+        if event_type == "subagent.tool":
             st["seq"] += 1
-            tool = {
-                "name": str(payload.get("tool_name") or "tool"),
-                "tool_id": f"submirror:{child_key}:{st['seq']}",
-                "args": {},
-            }
+            tool = {"name": str(payload.get("tool_name") or "tool"),
+                    "tool_id": f"submirror:{child_key}:{st['seq']}", "args": {}}
             if preview := str(payload.get("tool_preview") or payload.get("text") or ""):
                 tool["preview"] = preview
             st["open_tool"] = tool
             _emit("tool.start", csid, tool)
-        elif event_type == "subagent.complete":
-            if st["open_tool"]:
-                _emit("tool.complete", csid, st["open_tool"])
+        else:
             summary = str(payload.get("summary") or payload.get("text") or "")
             _emit("message.complete", csid, {"text": summary})
             _child_mirrors.pop(child_key, None)
@@ -100,8 +97,7 @@ def _agent_cbs(sid: str) -> dict:
         "tool_start_callback": lambda tc_id, name, args: _on_tool_start(sid, tc_id, name, args),
         "tool_complete_callback": lambda tc_id, name, args, result: _on_tool_complete(sid, tc_id, name, args, result),
         "tool_progress_callback": lambda event_type, name=None, preview=None, args=None, **kwargs: _on_tool_progress(
-            sid, event_type, name, preview, args, **kwargs
-        ),
+            sid, event_type, name, preview, args, **kwargs),
         "tool_gen_callback": lambda name: _tool_progress_enabled(sid) and _emit("tool.generating", sid, {"name": name}),
         "thinking_callback": lambda text: _emit("thinking.delta", sid, {"text": text}),
         # Affection reaction (ily / <3 / good bot) → hearts; core-detected so TUI and desktop share it.
@@ -112,14 +108,12 @@ def _agent_cbs(sid: str) -> dict:
         "status_callback": lambda kind, text=None: _status_update(sid, str(kind), None if text is None else str(text)),
         # Credits/notice spine: AgentNotice → notification.show; recovery clear → notification.clear.
         "notice_callback": lambda n: _emit(
-            "notification.show",
-            sid,
+            "notification.show", sid,
             {"text": n.text, "level": n.level, "kind": n.kind, "ttl_ms": n.ttl_ms, "key": n.key, "id": n.id},
         ),
         "notice_clear_callback": lambda key: _emit("notification.clear", sid, {"key": key}),
         "clarify_callback": lambda q, c, multi_select=False, questions=None: (
-            _clarify_block(sid, q, c, multi_select=multi_select, questions=questions)
-        ),
+            _clarify_block(sid, q, c, multi_select=multi_select, questions=questions)),
         "read_terminal_callback": _read_block("terminal.read.request", 30),
         "read_preview_callback": _read_block("preview.read.request", 45),
         # drive_preview / annotate_preview (desktop GUI): renderer drives the preview webview and
@@ -133,16 +127,14 @@ def _agent_cbs(sid: str) -> dict:
             "mcp.setup.request", sid, {"server": server, "action": action, "reason": reason}, timeout=600
         ),
         # tour (desktop GUI): renderer drives driver.js and answers tour.respond.
-        "tour_callback": lambda payload: _tour_request(sid, payload),
-    }
+        "tour_callback": lambda payload: _tour_request(sid, payload)}
 
     # Interim assistant commentary (text alongside tool calls). Gated on
     # display.interim_assistant_messages (default true); _run_prompt_submit overwrites
     # it per turn and clears it in its finally so a stale closure can't fire.
     if _load_interim_assistant_messages():
         callbacks["interim_assistant_callback"] = lambda text, *, already_streamed=False: _emit(
-            "message.interim", sid, {"text": str(text), "already_streamed": bool(already_streamed)}
-        )
+            "message.interim", sid, {"text": str(text), "already_streamed": bool(already_streamed)})
 
     return callbacks
 
@@ -155,15 +147,12 @@ def _apply_project_workspace(task_id: str, path: str, _name: str = "") -> None:
         return
     # task_id is the durable session_key; _sessions (and desktop event routing) key by sid.
     key = str(task_id or "")
-    sid, session = "", None
     with _sessions_lock:
-        if key in _sessions:
-            sid, session = key, _sessions[key]
-        else:
-            for cand_sid, cand in _sessions.items():
-                if cand.get("session_key") == key or getattr(cand.get("agent"), "session_id", None) == key:
-                    sid, session = cand_sid, cand
-                    break
+        sid, session = (key, _sessions[key]) if key in _sessions else next(
+            ((s, c) for s, c in _sessions.items()
+             if c.get("session_key") == key or getattr(c.get("agent"), "session_id", None) == key),
+            ("", None),
+        )
     if session is None:
         return
     resolved = os.path.abspath(os.path.expanduser(str(path)))
@@ -176,13 +165,10 @@ def _apply_project_workspace(task_id: str, path: str, _name: str = "") -> None:
     _persist_session_cwd_and_schedule_git_meta(session, resolved)
     try:
         agent = session.get("agent")
-        if agent is not None:
-            info = _session_info(agent, session)
-        else:
-            info = {
-                "cwd": resolved, "branch": _git_branch_for_cwd(resolved),
-                "project": _project_info_for_cwd(resolved), "lazy": True,
-            }
+        info = _session_info(agent, session) if agent is not None else {
+            "cwd": resolved, "branch": _git_branch_for_cwd(resolved),
+            "project": _project_info_for_cwd(resolved), "lazy": True,
+        }
         _emit("session.info", sid, info)
     except Exception:
         logger.debug("failed to emit session.info after project workspace move", exc_info=True)
@@ -210,13 +196,6 @@ def _wire_callbacks(sid: str):
     set_secret_capture_callback(secret_cb)
 
 
-def _render_personality_prompt(value) -> str:
-    """Delegates to hermes_cli.personality (single owner of rendering)."""
-    from hermes_cli.personality import render_personality_prompt
-
-    return render_personality_prompt(value)
-
-
 def _available_personalities(cfg: dict | None = None) -> dict:
     """Built-ins + user overrides, via hermes_cli.personality (single owner)."""
     from hermes_cli.personality import available_personalities
@@ -230,7 +209,7 @@ def _validate_personality(value: str, cfg: dict | None = None) -> tuple[str, str
     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
+    from hermes_cli.personality import normalize_personality_name, render_personality_prompt
 
     name = normalize_personality_name(value)
     if not name:
@@ -239,7 +218,7 @@ def _validate_personality(value: str, cfg: dict | None = None) -> tuple[str, str
     if name not in personalities:
         names = ", ".join(f"`{n}`" for n in sorted(personalities))
         raise ValueError(f"Unknown personality: `{str(value).strip()}`.\n\nAvailable: `none`, {names}")
-    return name, _render_personality_prompt(personalities[name])
+    return name, render_personality_prompt(personalities[name])
 
 
 def _prompt_text(value) -> str:
@@ -250,8 +229,7 @@ def _prompt_text(value) -> str:
 
 
 def _apply_personality_to_session(
-    sid: str, session: dict, new_prompt: str, personality: str = ""
-) -> tuple[bool, dict | None]:
+    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
@@ -266,17 +244,14 @@ def _apply_personality_to_session(
     if not agent:
         return False, None
     agent.ephemeral_system_prompt = new_prompt or None
-    if new_prompt:
-        marker = (
-            "[System: The user has changed the assistant's personality. "
-            "From this point forward, adopt the following persona and respond "
-            f"accordingly: {new_prompt}]"
-        )
-    else:
-        marker = (
-            "[System: The user has cleared the personality overlay. "
-            "From this point forward, respond in your normal default style.]"
-        )
+    marker = (
+        "[System: The user has changed the assistant's personality. "
+        "From this point forward, adopt the following persona and respond "
+        f"accordingly: {new_prompt}]"
+        if new_prompt else
+        "[System: The user has cleared the personality overlay. "
+        "From this point forward, respond in your normal default style.]"
+    )
     # Like the model-switch marker: role=user so strict providers accept it
     # mid-conversation, but `display_kind` keeps it out of the
     # `truncate_before_user_ordinal` addressing space (untagged, every rewind would
@@ -298,9 +273,7 @@ def _cfg_max_turns(cfg: dict, default: int) -> int:
     raw = (cfg.get("agent") or {}).get("max_turns")
     if raw is None:
         raw = cfg.get("max_turns")
-    if raw is not None:
-        return _resolve_turn_limit(raw, default=default)
-    return default
+    return default if raw is None else _resolve_turn_limit(raw, default=default)
 
 
 def _parse_tui_skills_env() -> list[str]:
@@ -326,45 +299,45 @@ def _agent_fallback_model(agent):
     """Return an agent's fallback chain without rehydrating deliberately empty chains."""
     if hasattr(agent, "_fallback_chain"):
         return agent._fallback_chain or []
-    if hasattr(agent, "_fallback_model"):
-        return agent._fallback_model
-    return _load_fallback_model()
+    return agent._fallback_model if hasattr(agent, "_fallback_model") else _load_fallback_model()
 
 
 def _background_agent_kwargs(agent, task_id: str) -> dict:
     cfg = _load_cfg()
 
+    def g(name, default=None):
+        return getattr(agent, name, default)
+
     return {
-        "base_url": getattr(agent, "base_url", None) or None,
-        "api_key": getattr(agent, "api_key", None) or None,
-        "provider": getattr(agent, "provider", None) or None,
-        "api_mode": getattr(agent, "api_mode", None) or None,
-        "acp_command": getattr(agent, "acp_command", None) or None,
-        "acp_args": getattr(agent, "acp_args", None) or None,
-        "model": getattr(agent, "model", None) or _resolve_model(),
+        "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),
         # 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": getattr(agent, "enabled_toolsets", None) or _load_enabled_toolsets("tui"),
+        "enabled_toolsets": g("enabled_toolsets") or _load_enabled_toolsets("tui"),
         "quiet_mode": True,
         "verbose_logging": False,
-        "ephemeral_system_prompt": getattr(agent, "ephemeral_system_prompt", None) or None,
-        "providers_allowed": getattr(agent, "providers_allowed", None),
-        "providers_ignored": getattr(agent, "providers_ignored", None),
-        "providers_order": getattr(agent, "providers_order", None),
-        "provider_sort": getattr(agent, "provider_sort", None),
-        "provider_require_parameters": getattr(agent, "provider_require_parameters", False),
-        "provider_data_collection": getattr(agent, "provider_data_collection", None),
-        "openrouter_min_coding_score": getattr(agent, "openrouter_min_coding_score", None),
+        "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": getattr(agent, "reasoning_config", None)
-        or _load_reasoning_config(str(getattr(agent, "model", "") or "")),
-        "service_tier": getattr(agent, "service_tier", None) or _load_service_tier(),
-        "request_overrides": dict(getattr(agent, "request_overrides", {}) or {}),
+        "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),
-    }
+        "fallback_model": _agent_fallback_model(agent)}
 
 
 def _ephemeral_preview_agent_kwargs(agent, task_id: str) -> dict:
@@ -383,7 +356,6 @@ def _preview_restart_history(session: dict, max_messages: int = 24, max_tool_cha
             history = list(session.get("history", []) or [])
     except Exception:
         history = list(session.get("history", []) or [])
-
     if not history:
         return []
     start = max(0, len(history) - max_messages)
@@ -391,16 +363,14 @@ def _preview_restart_history(session: dict, max_messages: int = 24, max_tool_cha
         if history[idx].get("role") == "user":
             start = min(start, idx)
             break
-
     trimmed: list[dict] = []
     for msg in history[start:]:
         if not isinstance(msg, dict):
             continue
-        role = msg.get("role")
-        if role not in ("user", "assistant", "tool", "system"):
+        if msg.get("role") not in ("user", "assistant", "tool", "system"):
             continue
         copy = {k: v for k, v in msg.items() if k != "reasoning"}
-        if role == "tool":
+        if msg.get("role") == "tool":
             content = copy.get("content")
             if isinstance(content, str) and len(content) > max_tool_chars:
                 copy["content"] = content[:max_tool_chars] + f"\n... (truncated, original {len(content)} chars)"
@@ -416,8 +386,7 @@ def _preview_tool_result_preview(name: str, result: str) -> str:
     if not isinstance(data, dict):
         return ""
     if name == "terminal":
-        output = str(data.get("output") or "").strip()
-        if output:
+        if output := str(data.get("output") or "").strip():
             return output[-1200:]
         if data.get("session_id"):
             return f"Background process started: {data.get('session_id')}"
@@ -452,8 +421,7 @@ def _preview_restart_callbacks(parent: str, task_id: str) -> dict:
             progress(f"{event_type.replace('.', ' ')}: {name}")
 
     return {
-        "tool_start_callback": tool_start,
-        "tool_complete_callback": tool_complete,
+        "tool_start_callback": tool_start, "tool_complete_callback": tool_complete,
         "tool_progress_callback": tool_progress,
         "tool_gen_callback": lambda name: progress(f"Preparing {name}"),
         "status_callback": lambda kind, text=None: progress(text if text is not None else kind),
@@ -471,23 +439,19 @@ def _reset_session_agent(sid: str, session: dict) -> dict:
         for k in ("model_override", "create_reasoning_override", "create_service_tier_override", "one_turn_model_restore"):
             session.pop(k, None)
         new_agent = _make_agent(
-            sid,
-            session["session_key"],
-            session_id=session["session_key"],
+            sid, session["session_key"], session_id=session["session_key"],
             platform_override=_session_source(session),
-            context_cwd_is_launch_artifact=_context_cwd_is_launch_artifact(session),
-        )
+            context_cwd_is_launch_artifact=_context_cwd_is_launch_artifact(session))
     finally:
         _clear_session_context(tokens)
     session.update(
-        agent=new_agent, config_model_seen=_config_model_target(), attached_images=[], queued_prompt=None
-    )
-    session.pop("queued_prompts", None)
-    session["_queued_prompt_generation"] = int(session.get("_queued_prompt_generation", 0)) + 1
-    session.update(
+        agent=new_agent, config_model_seen=_config_model_target(), attached_images=[],
+        queued_prompt=None,
+        _queued_prompt_generation=int(session.get("_queued_prompt_generation", 0)) + 1,
         edit_snapshots={}, image_counter=0, running=False, show_reasoning=_load_show_reasoning(),
         tool_progress_mode=_load_tool_progress_mode(), tool_started_at={},
     )
+    session.pop("queued_prompts", None)
     with session["history_lock"]:
         session["history"] = []
         session["history_version"] = int(session.get("history_version", 0)) + 1
diff --git a/tui_gateway/entry.py b/tui_gateway/entry.py
index 16726e9113..16bdbb7362 100644
--- a/tui_gateway/entry.py
+++ b/tui_gateway/entry.py
@@ -1,10 +1,9 @@
 import os
 import sys
 
-# Stop a ``utils/`` (or ``proxy/``, ``ui/``) package in the launch directory
-# from shadowing Hermes's own top-level modules.  ``hermes_bootstrap`` lives at
-# the repo root (its name can't collide with a user package), so importing it
-# before the guard runs is safe.
+# Stop a ``utils/`` (or ``proxy/``, ``ui/``) package in the launch directory from
+# shadowing Hermes's own top-level modules. ``hermes_bootstrap`` lives at the repo
+# root (its name can't collide with a user package), so importing it first is safe.
 import hermes_bootstrap
 
 hermes_bootstrap.harden_import_path()
@@ -15,6 +14,7 @@ import signal
 import threading
 import time
 import traceback
+from contextlib import suppress
 
 from tui_gateway._env import env_float
 from tui_gateway._stdin_recovery import handle_spurious_eof
@@ -26,26 +26,20 @@ from tui_gateway.transport import TeeTransport
 
 logger = logging.getLogger(__name__)
 
-# Handle for a background MCP discovery thread spawned by THIS module.  Stays
-# None when discovery is delegated to the shared owner in hermes_cli.mcp_startup
-# (the current path); the wait/in-flight/join helpers consult both owners.
+# Discovery thread spawned by THIS module; None when delegated to the shared owner in
+# hermes_cli.mcp_startup (current path). The wait/in-flight/join helpers consult both.
 _mcp_discovery_thread = None
-
-# True once ensure_mcp_discovery_started found MCP servers configured and spawned
-# discovery through the shared owner.  Lets wait_for_mcp_discovery re-invoke the
-# idempotent spawn on later agent builds so the retry-after-zero-connected
-# allowance can fire (otherwise a first run that connected nothing latches the
-# process MCP-less).  A flag rather than a config re-probe so non-MCP sessions
-# never pay the tools.mcp_tool import on the per-agent-build wait path.
+# Set once MCP servers are found configured, so wait_for_mcp_discovery can re-invoke
+# the idempotent spawn on later builds (retry-after-zero-connected) without a config
+# re-probe — non-MCP sessions never pay the tools.mcp_tool import per build.
 _mcp_discovery_enabled = False
 
 
 def _install_sidecar_publisher() -> None:
     """Mirror every dispatcher emit to the dashboard sidebar via WS.
 
-    Activated by `HERMES_TUI_SIDECAR_URL`, set by the dashboard's ``/api/pty``
-    endpoint when a chat tab passes a ``channel`` query param.  Best-effort:
-    connect failure or runtime drop falls back to stdio-only.
+    Activated by `HERMES_TUI_SIDECAR_URL` (set by the dashboard's ``/api/pty``
+    endpoint). Best-effort: connect failure or runtime drop falls back to stdio-only.
     """
     url = os.environ.get("HERMES_TUI_SIDECAR_URL")
     if not url:
@@ -55,10 +49,9 @@ def _install_sidecar_publisher() -> None:
     server._stdio_transport = TeeTransport(server._stdio_transport, WsPublisherTransport(url))
 
 
-# Grace for orderly shutdown (atexit + finalisers) before ``os._exit(0)`` so a
-# wedged worker mid-flush can't strand the process.  1s covers the gateway's own
-# shutdown work; ``HERMES_TUI_GATEWAY_SHUTDOWN_GRACE_S`` overrides for slower
-# environments (a longer grace also means a longer wait on a real deadlock).
+# Grace for orderly shutdown before ``os._exit(0)`` so a worker wedged mid-flush can't
+# strand the process; ``HERMES_TUI_GATEWAY_SHUTDOWN_GRACE_S`` overrides (a longer grace
+# also means a longer wait on a real deadlock).
 _DEFAULT_SHUTDOWN_GRACE_S = 1.0
 
 
@@ -73,32 +66,24 @@ def _stamp() -> str:
 
 def _append_crash_log(header: str, dump=None) -> None:
     """Best-effort ``=== header ===`` entry in the crash log; ``dump(f)`` adds detail."""
-    try:
+    with suppress(Exception):
         os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True)
         with open(_CRASH_LOG, "a", encoding="utf-8") as f:
             f.write(f"\n=== {header} ===\n")
             if dump is not None:
                 dump(f)
-    except Exception:
-        pass
 
 
 def _log_signal(signum: int, frame) -> None:
     """Capture WHICH thread and WHERE a termination signal hit us, then exit.
 
-    SIG_DFL for SIGPIPE kills the process silently the instant a background
-    thread (TTS, beep, voice status) writes to a stdout the TUI stopped reading,
-    leaving no trace in the crash log.  ``sys.exit(0)`` alone used to race the
-    worker pool — a thread holding ``_stdout_lock`` mid-flush blocks interpreter
-    shutdown indefinitely — so we log all thread stacks, give the process the
-    configured grace to drain, and fall back to ``os._exit(0)``.
+    ``sys.exit(0)`` alone raced the worker pool — a thread holding ``_stdout_lock``
+    mid-flush blocks interpreter shutdown indefinitely — so log all thread stacks,
+    give the configured grace to drain, then ``os._exit(0)``.
     """
     # SIGPIPE/SIGHUP don't exist on Windows — only look up attributes present.
-    names = {
-        int(sig): attr
-        for attr in ("SIGPIPE", "SIGTERM", "SIGHUP", "SIGINT", "SIGBREAK")
-        if (sig := getattr(signal, attr, None)) is not None
-    }
+    names = {int(sig): attr for attr in ("SIGPIPE", "SIGTERM", "SIGHUP", "SIGINT", "SIGBREAK")
+             if (sig := getattr(signal, attr, None)) is not None}
     name = names.get(signum, f"signal {signum}")
 
     def _dump(f):
@@ -118,16 +103,13 @@ def _log_signal(signum: int, frame) -> None:
     timer = threading.Timer(_shutdown_grace_seconds(), lambda: os._exit(0))
     timer.daemon = True
     timer.start()
-
-    # The atexit handler (_shutdown_sessions) can be blocked past the grace
-    # window by a worker holding the GIL/_stdout_lock; finalize explicitly so
-    # unpersisted messages reach state.db before the hard-exit timer fires.
-    try:
+    # The atexit handler (_shutdown_sessions) can be blocked past the grace window
+    # by a worker holding the GIL/_stdout_lock; finalize explicitly so unpersisted
+    # messages reach state.db before the hard-exit timer fires.
+    with suppress(Exception):
         from tui_gateway.server import _shutdown_sessions
 
         _shutdown_sessions()
-    except Exception:
-        pass
 
     # Unwind the main thread so atexit + finalisers run inside the grace window;
     # the daemon timer is the safety net if that unwind hangs.
@@ -138,31 +120,24 @@ def _install_signal(signame, handler):
     """Install a signal handler if legal in this thread and platform.
 
     signal.signal() raises ValueError outside the main thread; skip silently so a
-    worker-thread first import (Desktop build path: server._build does ``from
-    tui_gateway.entry import ...``) doesn't abort.  Handlers are process-global,
-    so any main-thread import installs them for everyone.  Missing signals
-    (Windows: SIGPIPE/SIGHUP) are skipped too.
+    worker-thread first import (Desktop build path: server._build imports entry)
+    doesn't abort. Handlers are process-global, so any main-thread import installs
+    them for everyone. Missing signals (Windows: SIGPIPE/SIGHUP) are skipped too.
     """
-    if threading.current_thread() is not threading.main_thread():
-        return
     sig = getattr(signal, signame, None)
-    if sig is None:
+    if sig is None or threading.current_thread() is not threading.main_thread():
         return
-    try:
+    # Off the main thread despite the check, or handler rejected by the platform.
+    with suppress(ValueError, OSError, RuntimeError):
         signal.signal(sig, handler)
-    except (ValueError, OSError, RuntimeError):
-        # Off the main thread despite the check, or handler rejected by the
-        # platform — skip rather than crash the import.
-        pass
 
 
-# SIGPIPE: ignore, don't exit.  SIG_DFL killed the process silently whenever a
-# *background* thread wrote to a pipe the TUI had gone quiet on, even with the
-# main thread fine on stdin.  Ignoring lets the write raise BrokenPipeError
-# (write_json handles it with a clean sys.exit(0) + _log_exit) so the gateway
-# lives as long as the command pipe is readable.  Terminal signals route through
-# _log_signal so kills/hangups are diagnosable; SIGBREAK (Windows Ctrl+Break) is
-# the weaker SIGHUP equivalent.
+# SIGPIPE: ignore, don't exit. SIG_DFL killed the process silently whenever a
+# *background* thread (TTS, beep, voice status) wrote to a pipe the TUI had gone
+# quiet on. Ignoring lets the write raise BrokenPipeError (write_json handles it with
+# a clean sys.exit(0) + _log_exit) so the gateway lives as long as the command pipe
+# is readable. Terminal signals route through _log_signal so kills/hangups are
+# diagnosable; SIGBREAK (Windows Ctrl+Break) is the weaker SIGHUP.
 _install_signal("SIGPIPE", signal.SIG_IGN)
 _install_signal("SIGTERM", _log_signal)
 if hasattr(signal, "SIGHUP"):
@@ -173,12 +148,9 @@ _install_signal("SIGINT", signal.SIG_IGN)
 
 
 def _log_exit(reason: str) -> None:
-    """Record why the gateway is shutting down.
-
-    Every exit path (startup/parse-error/response write fail, stdin EOF)
-    collapses into a silent sys.exit(0); without this trail the TUI shows
-    "gateway exited" with no clue WHICH broken pipe or message triggered it.
-    """
+    """Record why the gateway is shutting down: every exit path collapses into a
+    silent sys.exit(0), and without this trail the TUI shows "gateway exited" with
+    no clue WHICH broken pipe or message triggered it."""
     _append_crash_log(f"gateway exit · {_stamp()} · reason={reason}")
     print(f"[gateway-exit] {reason}", file=sys.stderr, flush=True)
 
@@ -186,13 +158,10 @@ def _log_exit(reason: str) -> None:
 def wait_for_mcp_discovery(timeout: "float | None" = None) -> None:
     """Block until background MCP discovery finishes, up to the resolved bound.
 
-    Discovery runs in a daemon thread so a slow/dead server can't freeze
-    ``gateway.ready``, but the agent snapshots its tool list ONCE at build time.
-    Joining with a bounded timeout before the first build lets already-spawning
-    servers land (join returns the instant discovery completes, so no-MCP
-    startups pay ~0s) without re-introducing the startup hang.  The bound is
-    ``mcp_discovery_timeout`` from config (via ``hermes_cli.mcp_startup``);
-    ``timeout`` overrides it.
+    The agent snapshots its tool list ONCE at build time, so a bounded join before
+    the first build lets already-spawning servers land (no-MCP startups pay ~0s)
+    without re-introducing the startup hang. Bound: ``mcp_discovery_timeout`` from
+    config; ``timeout`` overrides it.
     """
     thread = _mcp_discovery_thread
     if thread is not None and thread.is_alive():
@@ -204,12 +173,11 @@ def wait_for_mcp_discovery(timeout: "float | None" = None) -> None:
             bound = timeout if timeout is not None else 0.75
         thread.join(timeout=bound)
         return
-    # Shared-owner path.  Re-invoke the idempotent spawn first: if the previous
-    # run connected zero servers, the retry allowance starts a fresh run instead
-    # of leaving the process latched MCP-less.  It runs under the CALLER's
-    # profile context (agent build binds the session profile's HERMES_HOME
-    # first), so a launch profile without mcp_servers doesn't starve selected
-    # profiles.  Gated so non-MCP sessions never pay the tools.mcp_tool import.
+    # 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.
+    # It runs under the CALLER's profile context (agent build binds the session
+    # profile's HERMES_HOME first), so a launch profile without mcp_servers doesn't
+    # starve selected profiles. Gated so non-MCP sessions skip the mcp_tool import.
     if not _mcp_discovery_enabled:
         return
     try:
@@ -218,24 +186,17 @@ 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)
-    try:
+    with suppress(Exception):
         from hermes_cli.mcp_startup import wait_for_mcp_discovery as _startup_wait
 
         _startup_wait(timeout)
-    except Exception:
-        pass
 
 
 def mcp_discovery_in_flight() -> bool:
-    """True if ANY background MCP discovery thread is still running.
-
-    The agent-build path uses this to schedule a late tool-snapshot refresh when
-    discovery didn't land within the bounded join.  There are two owners by
-    surface — the stdio ``hermes --tui`` thread here and the
-    ``hermes_cli.mcp_startup`` thread used by desktop/dashboard — and the
-    late-refresh scheduler imports this regardless of surface, so it MUST
-    consult both or slow MCP servers' tools never surface on desktop.
-    """
+    """True if ANY background MCP discovery thread is still running. Two owners by
+    surface (stdio thread here, ``hermes_cli.mcp_startup`` for desktop/dashboard);
+    the late-refresh scheduler calls this regardless of surface, so it MUST consult
+    both or slow MCP servers' tools never surface on desktop."""
     thread = _mcp_discovery_thread
     if thread is not None and thread.is_alive():
         return True
@@ -248,12 +209,9 @@ def mcp_discovery_in_flight() -> bool:
 
 
 def join_mcp_discovery(timeout: float | None = None) -> bool:
-    """Join both discovery owners; True once neither is alive.
-
-    Unlike ``wait_for_mcp_discovery`` this accepts an unbounded/long wait and
-    reports the outcome (for the off-critical-path late-refresh waiter).
-    ``timeout`` bounds EACH join: entry thread first, then the shared owner.
-    """
+    """Join both discovery owners; True once neither is alive. Unlike
+    ``wait_for_mcp_discovery`` this accepts an unbounded wait (off-critical-path
+    late-refresh waiter); ``timeout`` bounds EACH join, entry thread first."""
     entry_done = True
     thread = _mcp_discovery_thread
     if thread is not None:
@@ -282,13 +240,12 @@ def _has_configured_mcp_servers() -> bool:
 def ensure_mcp_discovery_started() -> None:
     """Start background MCP discovery for the current profile context, once.
 
-    ``main()`` calls this for the stdio path.  WebSocket/Desktop entrypoints skip
-    ``main()``, so ``server._start_agent_build`` also calls it AFTER binding the
-    session profile's HERMES_HOME — the shared owner captures the caller's
-    context-local override into the discovery thread, 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.  Limitation: MCP registration is process-global, so the FIRST
+    ``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.
     """
     global _mcp_discovery_enabled
@@ -304,30 +261,33 @@ def ensure_mcp_discovery_started() -> None:
         logger.warning("Background MCP tool discovery failed to start", exc_info=True)
 
 
+def _write_or_exit(payload: dict, reason: str) -> None:
+    if not write_json(payload):
+        _log_exit(reason)
+        sys.exit(0)
+
+
 def main():
     _install_sidecar_publisher()
 
-    # Heartbeat row lets the orphan sweep tell "live but idle backend" from
-    # "truly orphaned"; must run BEFORE the sweep so it sees our row.
-    try:
-        server._start_backend_heartbeat_refresher()
-    except Exception:
-        logger.warning("backend heartbeat refresher start failed", exc_info=True)
-
-    # One-time sweep of rows orphaned by a previous gateway process (the
-    # in-process reap timer dies with the process).  Once-per-process and
-    # config-gated, so the handle_ws call site is a no-op when this ran.
-    try:
-        server._schedule_startup_orphan_sweep()
-    except Exception:
-        logger.warning("startup orphan sweep scheduling failed", exc_info=True)
+    # Heartbeat row lets the orphan sweep tell "live but idle backend" from "truly
+    # orphaned"; must run BEFORE the sweep so it sees our row. The sweep itself is
+    # once-per-process and config-gated (the handle_ws call site becomes a no-op).
+    for start, what in (
+        (server._start_backend_heartbeat_refresher, "backend heartbeat refresher start"),
+        (server._schedule_startup_orphan_sweep, "startup orphan sweep scheduling"),
+    ):
+        try:
+            start()
+        except Exception:
+            logger.warning("%s failed", what, exc_info=True)
 
     # Backgrounded so a dead MCP server (~7s of connect retries) can't freeze
-    # startup; _make_agent briefly joins it (wait_for_mcp_discovery).  The config
+    # startup; _make_agent briefly joins it (wait_for_mcp_discovery). The config
     # gate inside keeps the ~200ms MCP SDK import off the no-mcp_servers path.
     ensure_mcp_discovery_started()
 
-    if not write_json({
+    _write_or_exit({
         "jsonrpc": "2.0",
         "method": "event",
         "params": {
@@ -335,21 +295,17 @@ def main():
             # change_events: clients demote legacy polls (see tui_gateway/ws.py).
             # replay_epoch: WS restart detection; the stdio TUI ignores it.
             "payload": {
-                "skin": resolve_skin(),
-                "change_events": True,
-                "replay_epoch": replay_epoch(),
+                "skin": resolve_skin(), "change_events": True, "replay_epoch": replay_epoch(),
             },
         },
-    }):
-        _log_exit("startup write failed (broken stdout pipe before first event)")
-        sys.exit(0)
+    }, "startup write failed (broken stdout pipe before first event)")
 
     # Live-apply skins Hermes activates mid-conversation.
     server._ensure_skin_watcher()
 
     # Warm the /model picker's provider-models cache during this idle window
     # (mirrors the classic CLI loop); otherwise the first /model open blocks on
-    # serial /v1/models fetches.  Fire-and-forget, once-per-process.
+    # serial /v1/models fetches. Fire-and-forget, once-per-process.
     try:
         from hermes_cli.model_switch import prewarm_picker_cache_async
         prewarm_picker_cache_async()
@@ -371,16 +327,16 @@ def main():
         try:
             req = json.loads(line)
         except json.JSONDecodeError:
-            if not write_json({"jsonrpc": "2.0", "error": {"code": -32700, "message": "parse error"}, "id": None}):
-                _log_exit("parse-error-response write failed (broken stdout pipe)")
-                sys.exit(0)
+            _write_or_exit(
+                {"jsonrpc": "2.0", "error": {"code": -32700, "message": "parse error"}, "id": None},
+                "parse-error-response write failed (broken stdout pipe)")
             continue
 
         method = req.get("method") if isinstance(req, dict) else None
         resp = dispatch(req)
-        if resp is not None and not write_json(resp):
-            _log_exit(f"response write failed for method={method!r} (broken stdout pipe)")
-            sys.exit(0)
+        if resp is not None:
+            _write_or_exit(
+                resp, f"response write failed for method={method!r} (broken stdout pipe)")
 
 
 if __name__ == "__main__":
diff --git a/tui_gateway/mcp_oauth_sessions.py b/tui_gateway/mcp_oauth_sessions.py
index 25011a6dc3..d23845d057 100644
--- a/tui_gateway/mcp_oauth_sessions.py
+++ b/tui_gateway/mcp_oauth_sessions.py
@@ -1,28 +1,14 @@
 """Session-backed MCP OAuth flows for the gateway (mcp.servers.oauth.*).
 
-Mirrors the *provider* OAuth model used by the dashboard rather than the
-FastAPI-request-coupled MCP dashboard flow: ``start`` kicks off a background
+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 for that server
-in that profile.
-
-No OAuth logic is reimplemented here — the token machinery is the same one
-``hermes mcp login`` uses (``_probe_single_server`` under
-``force_interactive_oauth``).  ``DashboardOAuthFlow`` is reused verbatim as the
-thread-safe bridge (``publish_authorization_url`` / ``deliver_callback``); the
-only new piece is a tiny loopback HTTP listener on ``127.0.0.1:<port>/callback``
-that feeds ``deliver_callback`` instead of a FastAPI route.
-
-Client contract: ``start(profile, name)`` → open ``auth_url`` in the browser →
-poll until ``status`` is ``approved`` (tokens persisted) or ``error``.
-
-Remote-backend variant: when the desktop app runs on a DIFFERENT machine than
-the gateway, a gateway-side ``127.0.0.1`` listener is unreachable from the
-user's browser.  The client binds its OWN loopback listener, passes its
-``client_redirect_uri`` to ``start``, and relays the redirect back via
-``deliver_callback_flow``.  State verification stays server-side in
-``DashboardOAuthFlow.deliver_callback`` — a relayed code with the wrong
-``state`` is rejected exactly like a forged loopback hit.
+``{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.
 """
 
 from __future__ import annotations
@@ -31,6 +17,7 @@ import http.server
 import secrets
 import threading
 import time
+from contextlib import suppress
 from pathlib import Path
 from typing import Any, Dict, Optional
 from urllib.parse import parse_qs, urlparse
@@ -49,11 +36,8 @@ def _gc_sessions() -> None:
     """Drop expired sessions. Called opportunistically on start."""
     cutoff = time.time() - _SESSION_TTL_SECONDS
     with _sessions_lock:
-        stale = [sid for sid, rec in _sessions.items() if rec["created_at"] < cutoff]
-        for sid in stale:
-            rec = _sessions.pop(sid, None)
-            if rec is not None:
-                _shutdown_listener(rec)
+        for sid in [sid for sid, rec in _sessions.items() if rec["created_at"] < cutoff]:
+            _shutdown_listener(_sessions.pop(sid))
 
 
 def _shutdown_listener(rec: Dict[str, Any]) -> None:
@@ -61,30 +45,18 @@ def _shutdown_listener(rec: Dict[str, Any]) -> None:
     if server is None:
         return
     for stop in (server.shutdown, server.server_close):
-        try:
+        with suppress(Exception):
             stop()
-        except Exception:
-            pass
     rec["httpd"] = None
 
 
 def _validate_client_redirect_uri(uri: str) -> str:
-    """Validate a client-supplied loopback redirect URI.
-
-    Only plain-http loopback URLs (``http://127.0.0.1:<port>/...`` or
-    ``localhost``) are accepted, per RFC 8252 native-app rules — anything else
-    is rejected so the gateway can't pin an attacker-controlled redirect into a
-    DCR registration.
-    """
+    """Accept only plain-http loopback URLs (RFC 8252 native-app rules) so the
+    gateway can't pin an attacker-controlled redirect into a DCR registration."""
     parsed = urlparse(str(uri or "").strip())
     host = (parsed.hostname or "").lower()
-    if (
-        parsed.scheme != "http"
-        or host not in ("127.0.0.1", "localhost", "::1")
-        or not parsed.port
-        or parsed.username is not None
-        or parsed.password is not None
-    ):
+    if (parsed.scheme != "http" or host not in ("127.0.0.1", "localhost", "::1") or not parsed.port
+            or parsed.username is not None or parsed.password is not None):
         raise ValueError(
             "client_redirect_uri must be a loopback http URL like "
             "http://127.0.0.1:<port>/callback"
@@ -93,12 +65,9 @@ def _validate_client_redirect_uri(uri: str) -> str:
 
 
 def _start_loopback_listener(flow) -> "http.server.HTTPServer":
-    """Bind a loopback callback listener that feeds the flow's deliver_callback.
-
-    Returns the HTTPServer already serving on a daemon thread; the caller reads
-    the bound port off ``server_address`` to pin ``flow.redirect_uri`` BEFORE the
-    worker starts the flow (the redirect URI is fixed at authorization).
-    """
+    """Bind a loopback callback listener feeding ``flow.deliver_callback``; returns the
+    HTTPServer already serving on a daemon thread. The caller pins ``flow.redirect_uri``
+    from ``server_address`` BEFORE the worker starts (fixed at authorization)."""
 
     class _Handler(http.server.BaseHTTPRequestHandler):
         def do_GET(self):  # noqa: N802 — stdlib naming
@@ -119,49 +88,32 @@ def _start_loopback_listener(flow) -> "http.server.HTTPServer":
             self.send_response(status)
             self.send_header("Content-Type", "text/html; charset=utf-8")
             self.end_headers()
-            try:
+            with suppress(Exception):
                 self.wfile.write(body)
-            except Exception:
-                pass
 
         def log_message(self, *_a):  # silence stdlib request logging
             return
 
     httpd = http.server.HTTPServer(("127.0.0.1", 0), _Handler)
     threading.Thread(
-        target=httpd.serve_forever,
-        kwargs={"poll_interval": 0.5},
-        daemon=True,
-        name=f"mcp-oauth-cb-{flow.server_name}",
-    ).start()
+        target=httpd.serve_forever, kwargs={"poll_interval": 0.5}, daemon=True,
+        name=f"mcp-oauth-cb-{flow.server_name}").start()
     return httpd
 
 
 def _worker(session_id: str, hermes_home: str, server_name: str, cfg: dict, reconnect_live: bool) -> None:
-    """Drive the interactive MCP OAuth probe under the shared dashboard bridge.
-
-    Same HERMES_HOME override + secret-scope + force_interactive_oauth +
-    dashboard_oauth_flow wrapping around ``_probe_single_server`` as
-    ``web_server._run_dashboard_mcp_oauth``, keyed to our session record.  On
-    success the token file exists on disk and the server config is (re)saved
-    into the profile's config.yaml; on failure the prior token/manager state is
-    restored.
-    """
-    from hermes_cli.mcp_config import (
-        _oauth_tokens_present,
-        _probe_single_server,
-        _save_mcp_server,
-    )
+    """Drive the interactive MCP OAuth probe under the shared dashboard bridge (same
+    wrapping as ``web_server._run_dashboard_mcp_oauth``). On success the token file
+    exists and the server config is (re)saved; on failure the prior token/manager
+    state is restored."""
+    from hermes_cli.mcp_config import _oauth_tokens_present, _probe_single_server, _save_mcp_server
     from hermes_constants import reset_hermes_home_override, set_hermes_home_override
 
     rec = _sessions.get(session_id)
     flow = rec["flow"] if rec else None
     try:
         from agent.secret_scope import (
-            build_profile_secret_scope,
-            reset_secret_scope,
-            set_secret_scope,
-        )
+            build_profile_secret_scope, reset_secret_scope, set_secret_scope)
         from tools.mcp_dashboard_oauth import dashboard_oauth_flow
         from tools.mcp_oauth import force_interactive_oauth
         from tools.mcp_oauth_manager import get_manager
@@ -178,16 +130,12 @@ def _worker(session_id: str, hermes_home: str, server_name: str, cfg: dict, reco
                 previous_entry = None
                 try:
                     previous_entry = manager.remove(server_name, hermes_home=hermes_home)
-                    tools = _probe_single_server(
-                        server_name,
-                        cfg,
-                        connect_timeout=max(float(cfg.get("connect_timeout", 0) or 0), 315),
-                    )
+                    timeout = max(float(cfg.get("connect_timeout", 0) or 0), 315)
+                    tools = _probe_single_server(server_name, cfg, connect_timeout=timeout)
                     if not _oauth_tokens_present(server_name):
                         raise RuntimeError(
                             "The server responded, but no OAuth token was obtained — "
-                            "this provider may require a manually-registered OAuth client."
-                        )
+                            "this provider may require a manually-registered OAuth client.")
                     _save_mcp_server(server_name, cfg)
                     if flow is not None:
                         flow.tools = [{"name": t, "description": d} for t, d in tools]
@@ -205,14 +153,12 @@ def _worker(session_id: str, hermes_home: str, server_name: str, cfg: dict, reco
             reset_hermes_home_override(home_token)
     except Exception as exc:
         msg = str(exc)
-        try:
+        with suppress(Exception):
             from tools.mcp_oauth import humanize_oauth_registration_error
 
             msg = humanize_oauth_registration_error(
                 server_name, exc, server_url=cfg.get("url") if isinstance(cfg, dict) else None
             ) or msg
-        except Exception:
-            pass
         if flow is not None:
             flow.mark_error(msg)
     finally:
@@ -229,20 +175,11 @@ def start_flow(
     *,
     reconnect_live: bool = False,
     url_timeout: float = 30.0,
-    client_redirect_uri: Optional[str] = None,
-) -> Dict[str, Any]:
-    """Begin an MCP OAuth flow and return ``{session_id, auth_url, flow}``.
-
-    ``cfg`` is the server's resolved config (must have ``url`` and be
-    OAuth-capable); ``hermes_home`` the resolved profile home.  Blocks up to
-    ``url_timeout`` for the worker to publish the authorization URL.
-
-    ``client_redirect_uri`` (remote-backend variant): a loopback URL the CLIENT
-    hosts.  When set and valid, no gateway-side listener is bound — the OAuth
-    ``redirect_uri`` is pinned to the client's listener and the client relays
-    ``code``/``state`` via ``deliver_callback_flow``.  Invalid values raise
-    ``ValueError``.
-    """
+    client_redirect_uri: Optional[str] = None) -> Dict[str, Any]:
+    """Begin an MCP OAuth flow and return ``{session_id, auth_url, flow}``; blocks up
+    to ``url_timeout`` for the authorization URL. With ``client_redirect_uri`` (remote
+    backend; invalid values raise ``ValueError``) no gateway-side listener is bound and
+    the client relays ``code``/``state`` via ``deliver_callback_flow``."""
     from tools.mcp_dashboard_oauth import DashboardOAuthFlow
 
     if client_redirect_uri is not None:
@@ -259,39 +196,24 @@ def start_flow(
 
     session_id = secrets.token_urlsafe(24)
     flow = DashboardOAuthFlow(
-        flow_id=session_id,
-        server_name=server_name,
-        profile=None,
-        hermes_home=hermes_home,
+        flow_id=session_id, server_name=server_name, profile=None, hermes_home=hermes_home,
         redirect_uri="",  # set below once the loopback port is known
-        reconnect_live=reconnect_live,
-    )
-    if client_redirect_uri:
-        # Client hosts the callback listener; a 127.0.0.1 port here would be
-        # unreachable from the user's browser anyway.
-        httpd = None
-        flow.redirect_uri = client_redirect_uri
-    else:
-        httpd = _start_loopback_listener(flow)
-        flow.redirect_uri = f"http://127.0.0.1:{httpd.server_address[1]}/callback"
+        reconnect_live=reconnect_live)
+    # Client-hosted listener: a 127.0.0.1 port here would be unreachable from the browser.
+    httpd = None if client_redirect_uri else _start_loopback_listener(flow)
+    flow.redirect_uri = (
+        client_redirect_uri or f"http://127.0.0.1:{httpd.server_address[1]}/callback")
 
     rec = {
-        "session_id": session_id,
-        "server_name": server_name,
-        "hermes_home": hermes_home,
-        "flow": flow,
-        "httpd": httpd,
-        "created_at": time.time(),
+        "session_id": session_id, "server_name": server_name, "hermes_home": hermes_home,
+        "flow": flow, "httpd": httpd, "created_at": time.time(),
     }
     with _sessions_lock:
         _sessions[session_id] = rec
 
     threading.Thread(
-        target=_worker,
-        args=(session_id, hermes_home, server_name, dict(cfg), reconnect_live),
-        daemon=True,
-        name=f"mcp-oauth-{server_name}",
-    ).start()
+        target=_worker, args=(session_id, hermes_home, server_name, dict(cfg), reconnect_live),
+        daemon=True, name=f"mcp-oauth-{server_name}").start()
 
     try:
         auth_url = None
@@ -299,8 +221,7 @@ def start_flow(
         deadline = time.time() + url_timeout
         while time.time() < deadline:
             snap = flow.snapshot()
-            if snap.get("authorization_url"):
-                auth_url = snap["authorization_url"]
+            if auth_url := snap.get("authorization_url"):
                 break
             if snap.get("status") == "error":
                 raise RuntimeError(snap.get("error") or "MCP OAuth flow failed before authorization")
@@ -312,13 +233,9 @@ def start_flow(
         _shutdown_listener(rec)
         raise
 
-    return {
-        "session_id": session_id,
-        "auth_url": auth_url,
-        # Mirrors the provider-OAuth ``flow`` discriminator: open a URL then poll
-        # (no user_code to type, unlike device_code).
-        "flow": "pkce",
-    }
+    # ``flow`` mirrors the provider-OAuth discriminator: open a URL then poll
+    # (no user_code to type, unlike device_code).
+    return {"session_id": session_id, "auth_url": auth_url, "flow": "pkce"}
 
 
 def _lookup(session_id: str, server_name: str) -> "tuple[Dict[str, Any] | None, str | None]":
@@ -333,12 +250,9 @@ def _lookup(session_id: str, server_name: str) -> "tuple[Dict[str, Any] | None,
 
 
 def poll_flow(session_id: str, server_name: str) -> Dict[str, Any]:
-    """Poll a session's status → ``{status, error_message?, auth_url?, tools?}``.
-
-    ``status`` is ``pending`` | ``approved`` | ``error`` — the provider poll
-    vocabulary (``authorization_required`` from the bridge maps to ``pending``
-    since the client only needs to know whether to keep waiting).
-    """
+    """Poll a session → ``{status, error_message?, auth_url?, tools?}``; ``status``
+    is ``pending`` | ``approved`` | ``error`` (the bridge's ``authorization_required``
+    maps to ``pending`` — the client only needs to know whether to keep waiting)."""
     rec, err = _lookup(session_id, server_name)
     if rec is None:
         return {"status": "error", "error_message": err}
@@ -348,9 +262,7 @@ def poll_flow(session_id: str, server_name: str) -> Dict[str, Any]:
     raw = snap.get("status")
     status = raw if raw in ("approved", "error") else "pending"
     out: Dict[str, Any] = {
-        "session_id": session_id,
-        "status": status,
-        "error_message": snap.get("error"),
+        "session_id": session_id, "status": status, "error_message": snap.get("error"),
         "auth_url": snap.get("authorization_url"),
     }
     if status == "approved":
@@ -359,20 +271,13 @@ def poll_flow(session_id: str, server_name: str) -> Dict[str, Any]:
 
 
 def deliver_callback_flow(
-    session_id: str,
-    server_name: str,
-    *,
-    code: Optional[str],
-    state: Optional[str],
+    session_id: str, server_name: str, *, code: Optional[str], state: Optional[str],
     error: Optional[str] = None,
 ) -> Dict[str, Any]:
-    """Relay a client-captured OAuth redirect into a session's flow.
-
-    Remote-backend companion to ``start_flow(client_redirect_uri=...)``.
-    Security is unchanged from the gateway-listener path — the underlying
-    ``DashboardOAuthFlow.deliver_callback`` verifies ``state`` (constant-time)
-    and rejects replays.  Returns ``{ok: true}`` or ``{ok: false, error_message}``.
-    """
+    """Relay a client-captured OAuth redirect into a session's flow (remote-backend
+    companion to ``start_flow(client_redirect_uri=...)``). Security is unchanged:
+    ``DashboardOAuthFlow.deliver_callback`` verifies ``state`` (constant-time) and
+    rejects replays. Returns ``{ok: true}`` or ``{ok: false, error_message}``."""
     rec, err = _lookup(session_id, server_name)
     if rec is None:
         return {"ok": False, "error_message": err}
diff --git a/tui_gateway/methods_projects.py b/tui_gateway/methods_projects.py
index f244719aad..5ae43b5bd8 100644
--- a/tui_gateway/methods_projects.py
+++ b/tui_gateway/methods_projects.py
@@ -1,5 +1,4 @@
-"""Projects RPC surface: first-class, per-profile, multi-folder workspaces, plus repo
-discovery and the sidebar project tree.
+"""Projects RPC surface: per-profile multi-folder workspaces, repo discovery, sidebar tree.
 
 Bodies are rebound onto server.py's globals at install time (see
 method_ctx.bind_module), so they reference server.py globals bare.
@@ -27,8 +26,7 @@ def _projects_payload(conn) -> dict:
     from hermes_cli import projects_db as pdb
     return {
         "projects": [p.to_dict() for p in pdb.list_projects(conn, include_archived=True)],
-        "active_id": pdb.get_active_id(conn),
-    }
+        "active_id": pdb.get_active_id(conn)}
 
 
 def _projects_method(name: str):
@@ -65,6 +63,14 @@ 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 _path_param(params: dict) -> str:
+    return str(params.get("path") or "")
+
+
 @_projects_method("projects.list")
 def _(rid, params, pdb, conn) -> dict:
     return _ok(rid, _projects_payload(conn))
@@ -81,8 +87,7 @@ def _(rid, params, pdb, conn) -> dict:
         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"),
-    )
+        board_slug=params.get("board_slug"))
     if params.get("use"):
         pdb.set_active(conn, pid)
     proj = pdb.get_project(conn, pid)
@@ -94,33 +99,32 @@ 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 _ok(rid, {"project": pdb.get_project(conn, proj.id).to_dict()})
+        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, str(params.get("path") or ""), label=params.get("label"),
+        conn, proj.id, _path_param(params), label=params.get("label"),
         is_primary=bool(params.get("is_primary")),
     )
-    return _ok(rid, {"project": pdb.get_project(conn, proj.id).to_dict()})
+    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, str(params.get("path") or ""))
-    return _ok(rid, {"project": pdb.get_project(conn, proj.id).to_dict()})
+    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, str(params.get("path") or ""))
-    return _ok(rid, {"project": pdb.get_project(conn, proj.id).to_dict()})
+    pdb.set_primary(conn, proj.id, _path_param(params))
+    return _project_ok(rid, pdb, conn, proj.id)
 
 
 @_projects_method("projects.archive")
@@ -151,15 +155,13 @@ def _(rid, params, pdb, conn) -> dict:
 
 
 def _non_workspace_dirs() -> set[str]:
-    """Directories that are never a workspace, whichever tier proposes them.
+    """Directories that are never a workspace: ``/``, the user's home, and the dir
+    homes live in (``/home``, ``/Users``, ``C:\\Users``).
 
-    The filesystem root, the user's home, and the directory homes live in —
-    ``/home`` on Linux, ``/Users`` on macOS, ``C:\\Users`` on Windows. Both
-    POSIX spellings are excluded on every host because both are reachable as a
-    cwd anywhere: macOS ships an empty ``/home`` autofs stub, and a container or
-    remote shell hands back Linux paths. Promoting one of these mints a
-    catch-all project that swallows unplaced sessions, and ``/home`` in
-    particular renders as a second row reading "home" next to the Home bucket.
+    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.
     """
     home = os.path.realpath(os.path.expanduser("~"))
     candidates = (os.sep, home, os.path.dirname(home), "/home", "/Users")
@@ -167,10 +169,9 @@ def _non_workspace_dirs() -> set[str]:
 
 
 def _is_repo_junk(root: str) -> bool:
-    """A git root we never auto-surface as a project: a non-workspace dir (see
-    :func:`_non_workspace_dirs`) or anything under HERMES_HOME (~/.hermes by
-    default) — config/sessions/skills, not a workspace. User-created projects
-    pointing there are still honored."""
+    """A git root never auto-surfaced as a project: a non-workspace dir or anything
+    under HERMES_HOME (config/sessions/skills). User-created projects pointing
+    there are still honored."""
     if not root:
         return True
     from hermes_constants import get_hermes_home
@@ -179,18 +180,15 @@ def _is_repo_junk(root: str) -> bool:
     return (
         os.path.normcase(real) in _non_workspace_dirs()
         or real == hermes_home
-        or real.startswith(hermes_home + os.sep)
-    )
+        or real.startswith(hermes_home + os.sep))
 
 
 def _is_session_cwd_junk(cwd: str) -> bool:
-    """A non-git cwd that should stay in flat Recents rather than auto-group.
+    """A non-git cwd that stays in flat Recents rather than auto-grouping.
 
-    Unlike discovered git roots, an explicitly selected descendant of
-    HERMES_HOME may be an intentional prose/data workspace. The pre-Projects
-    desktop surfaced every such cwd, so exclude only the broad defaults that
-    would create catch-all projects: HERMES_HOME itself and the dirs in
-    :func:`_non_workspace_dirs`.
+    Unlike git roots, an explicitly selected DESCENDANT of HERMES_HOME may be an
+    intentional prose/data workspace (the pre-Projects desktop surfaced every
+    cwd), so only HERMES_HOME itself and ``_non_workspace_dirs`` are excluded.
     """
     if not cwd:
         return True
@@ -207,23 +205,21 @@ def _repo_discovery_policy(raw: dict | None = None) -> dict:
     source = raw if isinstance(raw, dict) else (_load_cfg().get("desktop") or {})
     if not isinstance(source, dict):
         source = {}
-    enabled = source.get("enabled", source.get("repo_scan_enabled", defaults["repo_scan_enabled"]))
-    roots = source.get("roots", source.get("repo_scan_roots", defaults["repo_scan_roots"]))
-    excludes = source.get(
-        "exclude_paths", source.get("repo_scan_exclude_paths", defaults["repo_scan_exclude_paths"]),
-    )
+
+    def _get(short: str, long: str):
+        return source.get(short, source.get(long, defaults[long]))
+
+    def _paths(values, long: str) -> list[str]:
+        if not isinstance(values, list):
+            return list(defaults[long])
+        return [v.strip() for v in values if isinstance(v, str) and v.strip()]
+
+    enabled = _get("enabled", "repo_scan_enabled")
     return {
         "enabled": enabled if isinstance(enabled, bool) else defaults["repo_scan_enabled"],
-        "roots": [value.strip() for value in roots if isinstance(value, str) and value.strip()]
-        if isinstance(roots, list)
-        else list(defaults["repo_scan_roots"]),
-        "exclude_paths": [
-            value.strip()
-            for value in excludes
-            if isinstance(value, str) and value.strip()
-        ]
-        if isinstance(excludes, list)
-        else list(defaults["repo_scan_exclude_paths"]),
+        "roots": _paths(_get("roots", "repo_scan_roots"), "repo_scan_roots"),
+        "exclude_paths": _paths(
+            _get("exclude_paths", "repo_scan_exclude_paths"), "repo_scan_exclude_paths"),
     }
 
 
@@ -239,36 +235,28 @@ def _repo_discovery_policy_key(policy: dict) -> str:
         return sorted(normalized)
     canonical = {
         "enabled": bool(policy["enabled"]), "roots": _paths(policy["roots"]),
-        "exclude_paths": _paths(policy["exclude_paths"]),
-    }
+        "exclude_paths": _paths(policy["exclude_paths"])}
     return json.dumps(canonical, sort_keys=True, separators=(",", ":"))
 
 
 def _repo_discovery_policy_is_default(policy: dict) -> bool:
     from hermes_cli.config import DEFAULT_CONFIG
     return _repo_discovery_policy_key(policy) == _repo_discovery_policy_key(
-        _repo_discovery_policy(DEFAULT_CONFIG["desktop"])
-    )
+        _repo_discovery_policy(DEFAULT_CONFIG["desktop"]))
 
 
 def _scan_discovered_repos_remote(conn, policy: dict) -> bool:
-    """Backend-side disk scan of the discovery policy roots.
+    """Backend-side disk scan of the discovery policy roots into the discovery cache.
 
-    The desktop's native repo scan only runs on the local filesystem. On a
-    remote gateway connection the host must scan its own disk so repos with
-    zero Hermes sessions still appear in the sidebar (#81723). Mirrors the
-    desktop's behavior: walk each root (bounded depth), find `.git`
-    directories, record (root, label) pairs 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.
 
-    Best-effort: any failure logs and leaves the cache untouched — the
-    session-derived repos from `_discover_repos_payload` still surface.
-
-    Returns True when the scan is authoritative (every root was walked to
-    completion without error and the per-scan cap was not hit). Only then may
-    the caller treat the result as a full replacement and pass ``replace=True``
-    to the cache write — a partial or errored scan must merge, never wipe, so
-    a failed remote refresh can't blank the previously cached repos into the
-    silent, unpopulated sidebar of #81723.
+    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.
     """
     from hermes_cli import projects_db as pdb
     roots = policy.get("roots") or []
@@ -292,36 +280,28 @@ def _scan_discovered_repos_remote(conn, policy: dict) -> bool:
                 if _is_excluded(dirpath):
                     dirnames[:] = []
                     continue
-                # A `.git` directory marks this directory as a repo root. Check
-                # BEFORE pruning hidden dirs — `.git` is itself hidden, so a
-                # prune-first order would drop it and never detect any repo.
+                # Check `.git` BEFORE pruning hidden dirs — `.git` is itself hidden.
                 if ".git" in dirnames:
-                    repo_root = dirpath
-                    if repo_root not in seen:
-                        seen.add(repo_root)
-                        pairs.append((repo_root, os.path.basename(repo_root)))
-                    # Don't descend into the repo's own .git to hunt nested repos.
-                    dirnames[:] = []
+                    if dirpath not in seen:
+                        seen.add(dirpath)
+                        pairs.append((dirpath, os.path.basename(dirpath)))
+                    dirnames[:] = []  # don't hunt nested repos inside a repo
                 else:
-                    # Not a repo: skip hidden dirs (e.g. .hermes) and node_modules.
                     dirnames[:] = [d for d in dirnames if not d.startswith(".") and d not in ("node_modules",)]
                 if len(pairs) >= 500:
                     break
         except Exception:
-            # A root that can't be walked yields no authoritative set — fall back
-            # to merging, never replacing, so the prior cache survives.
             authoritative = False
             logger.debug("discover_repos scan failed for root %s", root, exc_info=True)
         if len(pairs) >= 500:
-            # Cap hit means the walk didn't cover the full roots; the collected
-            # set must not be treated as the complete authoritative universe.
+            # Cap hit: the walk didn't cover the full roots, so this set is not
+            # the complete authoritative universe.
             authoritative = False
             break
     if pairs:
         try:
             pdb.record_discovered_repos(
-                conn, pairs, replace=authoritative, policy_key=_repo_discovery_policy_key(policy)
-            )
+                conn, pairs, replace=authoritative, policy_key=_repo_discovery_policy_key(policy))
         except Exception:
             logger.debug("discover_repos cache write failed", exc_info=True)
             authoritative = False
@@ -329,28 +309,20 @@ 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]:
+    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 (persisted by `projects.record_repos`) surfaces
-    repos even with zero hermes sessions. Session-derived roots cover repos
-    outside the scan roots. Both are junk-filtered (hermes home subtree + bare
-    home) and carry their session totals for the overview.
-
-    ``conn`` reuses an already-open projects.db connection (the tree path holds
-    one); ``backfill`` persists resolved roots back onto session rows — kept off
-    the per-turn tree path (grouping uses the live git resolver regardless) and
-    done only on the explicit discover/record refresh.
+    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.
     """
-    _is_junk = _is_repo_junk
     repos: dict[str, dict] = {}
 
     def _agg(root: str) -> dict:
         return repos.setdefault(root, {"root": root, "label": "", "sessions": 0, "last_active": 0.0})
 
-    # Session-derived roots (common repo root, folding worktrees; cached) +
-    # backfill the column so persisted git_repo_root matches the tree grouping.
     cwd_rows = list(db.distinct_session_cwds())
     # Warm the per-cwd git probes in parallel so a cold first paint doesn't
     # serialize one subprocess per distinct cwd before this loop reads the cache.
@@ -362,7 +334,7 @@ def _discover_repos_payload(
         if not root:
             continue
         cwd_to_root[cwd] = root
-        if _is_junk(root):
+        if _is_repo_junk(root):
             continue
         agg = _agg(root)
         agg["sessions"] += int(row.get("sessions") or 0)
@@ -373,26 +345,17 @@ def _discover_repos_payload(
         except Exception:
             logger.debug("failed to backfill repo roots", exc_info=True)
     if include_cached:
-        # Filesystem-scanned roots from the cache (may have zero sessions). Reuse
-        # the caller's projects.db connection when given, else a short-lived one.
+        # `last_seen` is scan time, not user activity — never fold it into
+        # `last_active` (made every scanned repo "just now").
         try:
             from hermes_cli import projects_db as pdb
-
-            def _read(c) -> None:
+            with (contextlib.nullcontext(conn) if conn is not None else pdb.connect_closing()) as c:
                 for entry in pdb.list_discovered_repos(c):
                     root = str(entry.get("root") or "")
-                    if not root or _is_junk(root):
-                        continue
-                    agg = _agg(root)
-                    if entry.get("label"):
-                        agg["label"] = entry["label"]
-                    # `last_seen` is scan time, not user activity; folding it
-                    # into `last_active` made every scanned repo "just now".
-            if conn is not None:
-                _read(conn)
-            else:
-                with pdb.connect_closing() as own:
-                    _read(own)
+                    if root and not _is_repo_junk(root):
+                        agg = _agg(root)
+                        if entry.get("label"):
+                            agg["label"] = entry["label"]
         except Exception:
             logger.debug("failed to read discovered repo cache", exc_info=True)
     out = sorted(repos.values(), key=lambda r: r["last_active"], reverse=True)
@@ -407,42 +370,23 @@ _PROJECT_TREE_EXCLUDED_SOURCES = ["cron", "kanban"]
 
 
 def _project_tree_row(r: dict) -> dict:
-    """Project a SessionDB row to the minimal shape the sidebar renders.
-
-    Keeps the fields the grouping needs (cwd / git_branch / git_repo_root) plus
-    everything ``SidebarSessionRow`` reads, and drops the heavy columns
-    (system_prompt, model_config, ...) so the tree payload stays lean.
-    """
-    return {
-        "id": r.get("id"),
-        "_lineage_root_id": r.get("_lineage_root_id"),
-        "_lineage_ids": r.get("_lineage_ids"),
-        # The sidebar nests branch/fork sessions under their parent
-        # (flattenSessionsWithBranches keys on this); without it, lane rows can't
-        # draw the └─ connector the flat Recents list shows.
-        "parent_session_id": r.get("parent_session_id"),
-        "title": r.get("title"),
-        "preview": r.get("preview"),
-        "started_at": r.get("started_at") or 0,
-        "ended_at": r.get("ended_at"),
-        "last_active": r.get("last_active") or r.get("started_at") or 0,
-        "source": r.get("source"),
-        "archived": bool(r.get("archived")),
-        "message_count": r.get("message_count") or 0,
-        "tool_call_count": r.get("tool_call_count") or 0,
-        "input_tokens": r.get("input_tokens") or 0,
-        "output_tokens": r.get("output_tokens") or 0,
-        # Cost is one of the fields SidebarSessionRow renders, so a lane row has
-        # to carry it too — without it, switching Show → cost filled in every
-        # figure in Recents and left the same sessions blank under a project.
-        "actual_cost_usd": r.get("actual_cost_usd"),
-        "estimated_cost_usd": r.get("estimated_cost_usd"),
-        "model": r.get("model"),
-        "is_active": False,
-        "cwd": r.get("cwd"),
-        "git_branch": r.get("git_branch"),
-        "git_repo_root": r.get("git_repo_root"),
-    }
+    """Project a SessionDB row to the minimal shape the sidebar renders: the
+    grouping fields (cwd/git_branch/git_repo_root) + everything ``SidebarSessionRow``
+    reads (parent_session_id for the └─ connector, cost for Show → cost), minus the
+    heavy columns."""
+    row = {k: r.get(k) for k in (
+        "id", "_lineage_root_id", "_lineage_ids", "parent_session_id", "title", "preview")}
+    row.update(
+        started_at=r.get("started_at") or 0, ended_at=r.get("ended_at"),
+        last_active=r.get("last_active") or r.get("started_at") or 0,
+        source=r.get("source"), archived=bool(r.get("archived")))
+    row.update({k: r.get(k) or 0 for k in (
+        "message_count", "tool_call_count", "input_tokens", "output_tokens")})
+    row.update(
+        actual_cost_usd=r.get("actual_cost_usd"), estimated_cost_usd=r.get("estimated_cost_usd"),
+        model=r.get("model"), is_active=False, cwd=r.get("cwd"), git_branch=r.get("git_branch"),
+        git_repo_root=r.get("git_repo_root"))
+    return row
 
 
 def _project_tree_inputs(
@@ -450,10 +394,9 @@ def _project_tree_inputs(
 ) -> 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 entered
-    view (drill-in) skips it entirely — it only needs the project it's showing,
-    which already has sessions — avoiding the distinct-cwd scan + git probes on
-    that per-turn path. One projects.db connection serves both reads.
+    ``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.
     """
     rows = db.list_sessions_rich(
         limit=session_limit,
@@ -463,15 +406,12 @@ def _project_tree_inputs(
         include_children=False,
         exclude_sources=_PROJECT_TREE_EXCLUDED_SOURCES,
         include_archived=False,
-        # `_project_tree_row` keeps ~18 fields and drops the rest, so selecting
-        # the system-prompt blob only to discard it costs tens of MB of B-tree
-        # reads per build on a long-lived database.
-        compact_rows=True,
-    )
+        # `_project_tree_row` drops the system-prompt blob; selecting it only to
+        # discard it costs tens of MB of B-tree reads per build on a big DB.
+        compact_rows=True)
     sessions = [_project_tree_row(r) for r in rows]
     # Parallel-warm the git cache so build_tree's resolver reads it instead of
-    # cold-probing each cwd in sequence (matters on the drill-in path, which
-    # skips the discovery warm-up below).
+    # cold-probing each cwd in sequence (matters on the drill-in path).
     git_probe.warm_roots(s["cwd"] for s in sessions if s.get("cwd"))
     from hermes_cli import projects_db as pdb
     policy = _repo_discovery_policy()
@@ -479,33 +419,26 @@ def _project_tree_inputs(
     with pdb.connect_closing() as conn:
         if include_discovered:
             pdb.reconcile_discovered_repos_policy(
-                conn, policy_key, preserve_unversioned=_repo_discovery_policy_is_default(policy),
-            )
+                conn, policy_key, preserve_unversioned=_repo_discovery_policy_is_default(policy))
         projects = [p.to_dict() for p in pdb.list_projects(conn)]
         active_id = pdb.get_active_id(conn)
         # backfill stays off the hot tree path — grouping uses the live resolver.
         discovered = (
             _discover_repos_payload(db, conn=conn, backfill=False, include_cached=policy["enabled"])
             if include_discovered
-            else []
-        )
+            else [])
     return sessions, projects, discovered, active_id
 
 
-# Per-build memo for `_dir_exists_cached`. Cleared at the top of every
-# `_build_project_tree`, so a dir created or deleted between sidebar refreshes
-# is seen on the next one.
+# Per-build memo for `_dir_exists_cached`; cleared at the top of every
+# `_build_project_tree` so a dir created/deleted between refreshes is seen.
 _DIR_EXISTS_CACHE: dict[str, bool] = {}
 
 
 def _dir_exists_cached(path: str) -> bool:
-    """``os.path.isdir`` for the project tree, memoized per build.
-
-    ``build_tree`` asks per SESSION, not per distinct path, so a power user with
-    hundreds of sessions across a handful of dirs would otherwise fire hundreds
-    of redundant stats on every sidebar open. The memo is per build, so a dir
-    created or deleted between refreshes is picked up on the next one.
-    """
+    """``os.path.isdir`` memoized per build — ``build_tree`` asks per SESSION, not
+    per distinct path, so hundreds of sessions in a few dirs would otherwise fire
+    hundreds of redundant stats per sidebar open."""
     hit = _DIR_EXISTS_CACHE.get(path)
     if hit is None:
         hit = os.path.isdir(path)
@@ -520,20 +453,16 @@ def _build_project_tree(
     from tui_gateway import project_tree
     _DIR_EXISTS_CACHE.clear()
     sessions, projects, discovered, active_id = _project_tree_inputs(
-        db, session_limit, include_discovered=include_discovered
-    )
-    # build_tree resolves every declared project folder and every discovered
-    # repo root too, and those paths are not session cwds — without this they
-    # are the one part of the build still probing git one directory at a time.
+        db, session_limit, include_discovered=include_discovered)
+    # build_tree also resolves every declared project folder and discovered repo
+    # root — not session cwds, so warm them too or they probe git one at a time.
     git_probe.warm_roots(
         [str(f.get("path") or "") for p in projects for f in (p.get("folders") or [])]
-        + [str(r.get("root") or "") for r in discovered]
-    )
+        + [str(r.get("root") or "") for r in discovered])
     tree = project_tree.build_tree(
         projects, sessions, discovered, _resolve_cwd_git, preview_limit=preview_limit,
         hydrate=hydrate, is_junk_root=_is_repo_junk, is_junk_cwd=_is_session_cwd_junk,
-        exists=_dir_exists_cached,
-    )
+        exists=_dir_exists_cached)
     return tree, active_id
 
 
diff --git a/tui_gateway/project_tree.py b/tui_gateway/project_tree.py
index bee7e9d6d5..e89ac5c835 100644
--- a/tui_gateway/project_tree.py
+++ b/tui_gateway/project_tree.py
@@ -1,23 +1,10 @@
-"""Authoritative project -> repo -> lane -> session tree builder.
+"""Authoritative project -> repo -> lane -> session tree builder (pure; git via ``resolve``).
 
-Single source of truth for how the desktop sidebar groups sessions. Pure (git
-resolution is injected via ``resolve``) so it is unit-testable and shared by the
-``projects.tree`` / ``projects.project_sessions`` RPCs.
-
-Emitted ids and lane keys must stay byte-compatible with the renderer's persisted
-state (pins, manual ordering, dismissal), which keys off these exact strings:
-
-  - explicit project id .......... ``p_<hex>`` (from projects.db)
-  - auto/discovered project id ... the repo root path
-  - home (no-project) bucket ..... ``__no_project__``
-  - repo node id ................. the repo root path
-  - main branch lane id .......... ``<repoRoot>::branch::<branch>`` (or ``::branch::``)
-  - kanban bucket lane id ........ ``<repoRoot>::kanban``
-  - linked worktree lane id ...... the worktree path
-
-Linked worktrees are folded under their MAIN repo via a git common-dir probe
-(``git rev-parse --show-toplevel`` returns the worktree's own root, which is why
-the old client-side grouping double-counted them).
+Emitted ids/lane keys must stay byte-compatible with the renderer's persisted state
+(pins, ordering, dismissal), which keys off: explicit project id ``p_<hex>``; auto
+project id / repo node id = repo root path; home bucket ``__no_project__``; main lane
+``<repoRoot>::branch::<branch>``; kanban lane ``<repoRoot>::kanban``; linked worktree
+lane = the worktree path. Linked worktrees fold under their MAIN repo (common-dir probe).
 """
 
 from __future__ import annotations
@@ -25,32 +12,26 @@ from __future__ import annotations
 import re
 from typing import Any, Callable, Optional
 
-# cwd -> ``{"repo_root", "worktree_root"}``: ``repo_root`` is the COMMON (main)
-# repo root shared across worktrees, ``worktree_root`` this cwd's own checkout
-# root. ``None`` when not in a git repo or unprobeable (remote backend).
+# cwd -> ``{"repo_root", "worktree_root"}`` (COMMON main root shared across worktrees /
+# this cwd's own checkout root); ``None`` when not in git or unprobeable (remote backend).
 Resolve = Callable[[str], Optional[dict]]
-
-# "does this directory still exist?" predicate. Defaults to True-for-everything
-# so callers that can't stat (remote backends) don't wrongly hide a project that
-# lives on the other host.
+# "does this directory still exist?"; defaults to always-True so callers that can't
+# stat (remote backends) don't hide a project living on the other host.
 Exists = Callable[[str], bool]
 
 # Only KANBAN-TASK worktrees (`<repo>/.worktrees/t_<hex>`, the id kanban_db mints)
 # collapse into one lane; user-named dirs under `.worktrees/` stay their own lanes.
 _KANBAN_DIR_RE = re.compile(r"^(.*[/\\]\.worktrees)[/\\]t_[0-9a-f]+[/\\]?$")
-_TRAILING_SEP_RE = re.compile(r"[/\\]+$")
 _TRUNK_BRANCHES = {"main", "master", "trunk", "develop"}
 DEFAULT_BRANCH_LABEL = "main"
 
-# Synthetic bucket for every session no project claimed (no cwd, bare home dir,
-# HERMES state, deleted workspace). The desktop labels it "Home"; the id/flag
-# name what the bucket MEANS since membership keys off them.
+# Synthetic bucket for every session no project claimed (no cwd, bare home, HERMES
+# state, deleted workspace); the id/flag name what it MEANS since membership keys off them.
 NO_PROJECT_ID = "__no_project__"
 NO_PROJECT_LABEL = "Home"
 
-# Sibling candidates tried when recovering a deleted worktree's parent repo
-# (``_probe_sibling_worktree``). Each miss costs a git probe; real suffixes are
-# one or two segments.
+# Sibling probes when recovering a deleted worktree's parent repo; each miss is a git
+# invocation and real suffixes are one or two segments.
 _MAX_SIBLING_PROBES = 4
 
 
@@ -75,11 +56,6 @@ def _kanban_lane_id(repo_root: str) -> str:
     return f"{repo_root}::kanban"
 
 
-# ---------------------------------------------------------------------------
-# Path helpers (match the TS segment logic so labels/ids line up)
-# ---------------------------------------------------------------------------
-
-
 def _segments(path: str) -> list[str]:
     return [s for s in re.split(r"[/\\]", (path or "").rstrip("/\\")) if s]
 
@@ -124,17 +100,17 @@ def kanban_worktree_dir(path: str) -> Optional[str]:
     return m.group(1) if m else None
 
 
-def _strip_trailing_sep(path: str) -> str:
-    return _TRAILING_SEP_RE.sub("", path or "")
-
-
 def _with_base_name(path: str, name: str) -> str:
-    return re.sub(r"[^/\\]+$", name, _strip_trailing_sep(path))
+    return re.sub(r"[^/\\]+$", name, (path or "").rstrip("/\\"))
 
 
 def _parent_dir(path: str) -> str:
     """The containing directory of ``path`` (``""`` once the root is passed)."""
-    return _strip_trailing_sep(_with_base_name(path, ""))
+    return _with_base_name(path, "").rstrip("/\\")
+
+
+def _field(row: dict, key: str) -> str:
+    return (row.get(key) or "").strip()
 
 
 def _branch_label(branch: str) -> str:
@@ -151,23 +127,13 @@ def _last_active(sessions: list[dict]) -> float:
     return max((_session_time(s) for s in sessions), default=0.0)
 
 
-# ---------------------------------------------------------------------------
-# Lane placement
-# ---------------------------------------------------------------------------
-
-
 def _placement(
     repo_root: str, lane_key: str, lane_label: str, lane_path: str, is_main: bool, is_kanban: bool
 ) -> dict:
     return {
-        "repo_key": repo_root,
-        "repo_label": base_name(repo_root) or repo_root,
-        "repo_path": repo_root,
-        "lane_key": lane_key,
-        "lane_label": lane_label,
-        "lane_path": lane_path,
-        "is_main": is_main,
-        "is_kanban": is_kanban,
+        "repo_key": repo_root, "repo_label": base_name(repo_root) or repo_root,
+        "lane_key": lane_key, "lane_label": lane_label, "lane_path": lane_path,
+        "is_main": is_main, "is_kanban": is_kanban,
     }
 
 
@@ -183,15 +149,13 @@ def _kanban_placement(repo_root: str, kanban_dir: str) -> dict:
 def _probe_sibling_worktree(cwd: str, resolve: Resolve) -> str:
     """The parent repo root of a deleted ``<repo>-<suffix>`` worktree, else ``""``.
 
-    A deleted dir can't be probed, so trim one ``-<segment>`` at a time off its
-    name and return the first sibling that resolves. The cwd is frequently a
-    SUBDIR of the deleted worktree (``<repo>-<suffix>/apps/desktop``) whose
-    basename shares nothing with the repo, so the trim is applied to each
-    ANCESTOR, deepest first — otherwise the dead path gets minted as its own
-    project. Probes are bounded in total (each is a git invocation).
+    A deleted dir can't be probed, so trim one ``-<segment>`` at a time off its name
+    and return the first sibling that resolves. The cwd is often a SUBDIR of the dead
+    worktree (``<repo>-<suffix>/apps/desktop``), so the trim runs on each ANCESTOR,
+    deepest first. Probes are bounded in total (each is a git invocation).
     """
     probes = 0
-    path = _strip_trailing_sep(cwd)
+    path = (cwd or "").rstrip("/\\")
     while path and probes < _MAX_SIBLING_PROBES:
         parts = base_name(path).split("-")
         for i in range(len(parts) - 1, 0, -1):
@@ -212,7 +176,7 @@ def _place_by_heuristic(path: str) -> Optional[dict]:
         return None
     kanban_dir = kanban_worktree_dir(path)
     if kanban_dir:
-        return _kanban_placement(_strip_trailing_sep(_with_base_name(kanban_dir, "")), kanban_dir)
+        return _kanban_placement(_parent_dir(kanban_dir), kanban_dir)
     m = re.match(r"^(.+)-wt-(.+)$", base)
     if m:
         return _placement(_with_base_name(path, m.group(1)), path, m.group(2), path, False, False)
@@ -249,42 +213,28 @@ def _place(cwd: str, branch: str, resolve: Optional[Resolve], persisted_root: st
 
 def _place_session(session: dict, resolve: Optional[Resolve]) -> Optional[dict]:
     """``_place`` for a session row; ``None`` when it has no cwd."""
-    cwd = (session.get("cwd") or "").strip()
+    cwd = _field(session, "cwd")
     if not cwd:
         return None
-    return _place(
-        cwd,
-        (session.get("git_branch") or "").strip(),
-        resolve,
-        (session.get("git_repo_root") or "").strip(),
-    )
+    return _place(cwd, _field(session, "git_branch"), resolve, _field(session, "git_repo_root"))
 
 
 def _session_repo_root(session: dict, resolve: Optional[Resolve]) -> str:
     """The COMMON repo root a session belongs to (folds linked worktrees)."""
-    cwd = (session.get("cwd") or "").strip()
+    cwd = _field(session, "cwd")
     if cwd and resolve:
         info = resolve(cwd)
         if info and info.get("repo_root"):
             return info["repo_root"]
-    return (session.get("git_repo_root") or "").strip()
-
-
-# ---------------------------------------------------------------------------
-# Ordering + label disambiguation (parity with the old client tree)
-# ---------------------------------------------------------------------------
+    return _field(session, "git_repo_root")
 
 
 def _lane_sort_key(group: dict) -> tuple:
     # Trunk pins to the top; the kanban aggregate sinks to the bottom; the rest
     # (branches + linked worktrees) sort by most-recent activity, then label.
     is_trunk = bool(group.get("isMain")) and group["label"].lower() in _TRUNK_BRANCHES
-    return (
-        0 if is_trunk else 1,
-        1 if group.get("isKanban") else 0,
-        -_last_active(group.get("sessions") or []),
-        group["label"].lower(),
-    )
+    return (0 if is_trunk else 1, 1 if group.get("isKanban") else 0,
+            -_last_active(group.get("sessions") or []), group["label"].lower())
 
 
 def _disambiguate_labels(items: list[dict]) -> None:
@@ -310,11 +260,6 @@ def _disambiguate_labels(items: list[dict]) -> None:
                 break
 
 
-# ---------------------------------------------------------------------------
-# Repo subtree assembly
-# ---------------------------------------------------------------------------
-
-
 def _repo_node(root: str, label: str) -> dict:
     return {"id": root, "label": label, "path": root, "groups": [], "sessionCount": 0}
 
@@ -328,26 +273,19 @@ def _build_repos(sessions: list[dict], resolve: Optional[Resolve], hydrate: bool
             continue
         lane_identity = _lane_key(placement["lane_key"])
         if lane_identity not in lanes:
-            lanes[lane_identity] = (
-                {
-                    "id": placement["lane_key"],
-                    "label": placement["lane_label"],
-                    "path": placement["lane_path"],
-                    "isMain": placement["is_main"],
-                    "isKanban": placement["is_kanban"],
-                    "sessions": [],
-                },
-                placement,
-            )
+            group = {
+                "id": placement["lane_key"], "label": placement["lane_label"],
+                "path": placement["lane_path"], "isMain": placement["is_main"],
+                "isKanban": placement["is_kanban"], "sessions": [],
+            }
+            lanes[lane_identity] = (group, placement)
         lanes[lane_identity][0]["sessions"].append(session)
 
     repos: dict[str, dict] = {}
     for group, placement in lanes.values():
         group["sessions"].sort(key=_session_time, reverse=True)
-        repo = repos.setdefault(
-            _path_key(placement["repo_key"]),
-            _repo_node(placement["repo_key"], placement["repo_label"]),
-        )
+        repo_key = placement["repo_key"]
+        repo = repos.setdefault(_path_key(repo_key), _repo_node(repo_key, placement["repo_label"]))
         repo["groups"].append(group)
         repo["sessionCount"] += len(group["sessions"])
 
@@ -365,21 +303,18 @@ def _build_repos(sessions: list[dict], resolve: Optional[Resolve], hydrate: bool
 
 
 def _seed_folder_repos(repos: list[dict], folders: list[dict], resolve: Optional[Resolve]) -> list[dict]:
-    """Ensure every declared project folder shows as a repo, even with 0 sessions.
-
-    Without it the desktop's entered-project view renders blank (early-returns on
-    no repos) and the optimistic live-session overlay has no lane to drop a
-    fresh session into until a full tree refresh. Folders already covered by a
-    session-derived repo (same git root) are left untouched.
-    """
+    """Ensure every declared project folder shows as a repo, even with 0 sessions:
+    otherwise the desktop's entered-project view renders blank and the optimistic
+    live-session overlay has no lane for a fresh session until a full refresh.
+    Folders already covered by a session-derived repo (same git root) are untouched."""
     seen = {_path_key(v) for repo in repos for v in (repo.get("id"), repo.get("path")) if v}
     seeded = list(repos)
     for folder in folders or []:
-        raw = (folder.get("path") or "").strip()
+        raw = _field(folder, "path")
         if not raw:
             continue
         info = resolve(raw) if resolve else None
-        root = (info or {}).get("repo_root") or _strip_trailing_sep(raw)
+        root = (info or {}).get("repo_root") or raw.rstrip("/\\")
         root_key = _path_key(root)
         if not root_key or root_key in seen:
             continue
@@ -390,28 +325,18 @@ def _seed_folder_repos(repos: list[dict], folders: list[dict], resolve: Optional
     return seeded
 
 
-# ---------------------------------------------------------------------------
-# Explicit-project ownership
-# ---------------------------------------------------------------------------
-
-
 class _FolderIndex:
-    """Normalized folder path -> (owning project, depth), so a session is matched
-    by walking its cwd's ancestors (O(path depth) lookups) instead of scanning
-    every project x folder per session."""
+    """Normalized folder path -> (owning project, depth): a session is matched by
+    walking its cwd's ancestors instead of scanning every project x folder."""
 
     def __init__(self, projects: list[dict]) -> None:
         self._by_path: dict[str, tuple[dict, int]] = {}
         for project in projects:
             for folder in project.get("folders") or []:
                 segs = _comparison_segments(folder.get("path") or "")
-                if not segs:
-                    continue
-                key = "/".join(segs)
                 # Deepest folder wins; ties keep the first project (scan order).
-                existing = self._by_path.get(key)
-                if existing is None or len(segs) > existing[1]:
-                    self._by_path[key] = (project, len(segs))
+                if segs and len(segs) > self._by_path.get("/".join(segs), (None, -1))[1]:
+                    self._by_path["/".join(segs)] = (project, len(segs))
 
     def match(self, target: str) -> tuple[Optional[dict], int]:
         """Owning project for ``target`` by longest ancestor folder, + its depth."""
@@ -424,22 +349,13 @@ class _FolderIndex:
 
 
 def _project_for_session(session: dict, index: _FolderIndex, resolve: Optional[Resolve]) -> Optional[dict]:
-    cwd = (session.get("cwd") or "").strip()
+    cwd = _field(session, "cwd")
     if not cwd:
         return None
     repo_root = _session_repo_root(session, resolve)
     candidates = [cwd, repo_root] if repo_root and repo_root != cwd else [cwd]
-    best, best_len = None, -1
-    for target in candidates:
-        match, length = index.match(target)
-        if match and length > best_len:
-            best, best_len = match, length
-    return best
-
-
-# ---------------------------------------------------------------------------
-# Public builder
-# ---------------------------------------------------------------------------
+    # Longest folder match wins; ties keep the cwd match (max() keeps the first maximum).
+    return max((index.match(t) for t in candidates), key=lambda hit: hit[1])[0]
 
 
 def _session_cost(session: dict) -> float:
@@ -451,37 +367,82 @@ def _session_cost(session: dict) -> float:
 
 
 def _project_node(
-    *,
-    pid: str,
-    label: str,
-    path: Optional[str],
-    repos: list[dict],
-    session_count: int,
-    last_active: float,
-    preview_sessions: list[dict],
-    sessions: Optional[list[dict]] = None,
-    color: Any = None,
-    icon: Any = None,
-    is_auto: bool = False,
-    is_no_project: bool = False,
+    pid: str, label: str, path: Optional[str], repos: list[dict], session_count: int,
+    last_active: float, preview_sessions: list[dict], sessions: Optional[list[dict]] = None,
+    **flags: Any,
 ) -> dict:
-    return {
-        "id": pid,
-        "label": label,
-        "path": path,
-        "color": color,
-        "icon": icon,
-        "isAuto": is_auto,
-        "isNoProject": is_no_project,
-        "sessionCount": session_count,
-        "lastActive": last_active,
+    """``flags`` overrides ``color`` / ``icon`` / ``isAuto`` / ``isNoProject`` (key order is
+    fixed by the defaults below — the renderer's wire shape)."""
+    node = {
+        "id": pid, "label": label, "path": path, "color": None, "icon": None,
+        "isAuto": False, "isNoProject": False,
+        "sessionCount": session_count, "lastActive": last_active,
         # Totals over the same sessions `sessionCount` counts, so a project header
         # adds up to what its rows show.
         "totalTokens": sum((s.get("input_tokens") or 0) + (s.get("output_tokens") or 0) for s in sessions or []),
         "totalCostUsd": sum(_session_cost(s) for s in sessions or []),
-        "repos": repos,
-        "previewSessions": preview_sessions,
+        "repos": repos, "previewSessions": preview_sessions,
     }
+    node.update(flags)
+    return node
+
+
+def _auto_buckets(
+    unowned: list[dict], resolve: Optional[Resolve], junk: Callable, junk_cwd: Callable,
+    exists: Callable,
+) -> tuple[dict[str, dict], list[dict]]:
+    """Group leftover sessions by auto-project root; the rest go to the Home bucket.
+    Prefer the common git root, then the session cwd for non-git workspaces (the
+    pre-Projects desktop grouped every cwd; dropping that flattens them into Recents)."""
+    by_auto_root: dict[str, dict] = {}
+    homeless: list[dict] = []
+
+    def _add_auto(root: str, session: dict) -> None:
+        key = _path_key(root)
+        if not key:
+            homeless.append(session)
+            return
+        by_auto_root.setdefault(key, {"root": root, "sessions": []})["sessions"].append(session)
+
+    for session in unowned:
+        root = _session_repo_root(session, resolve)
+        if root:
+            # A real git root uses the stricter repo policy; never reinterpret a
+            # filtered internal repo as a cwd-only project. A root no longer on
+            # disk is a stale persisted value and must not resurrect as a project.
+            if not junk(root) and exists(root):
+                _add_auto(root, session)
+            else:
+                homeless.append(session)
+            continue
+        cwd = _field(session, "cwd")
+        if not cwd or junk_cwd(cwd):
+            homeless.append(session)
+            continue
+        placement = _place_session(session, resolve)
+        # A placement that only echoes back an unresolvable cwd is the path-only
+        # heuristic guessing. If that dir is also gone from disk, promoting it
+        # mints a phantom project that can only be dismissed by hand -> Home.
+        if placement and exists(placement["repo_key"]):
+            _add_auto(placement["repo_key"], session)
+        else:
+            homeless.append(session)
+    return by_auto_root, homeless
+
+
+def _home_project(homeless: list[dict], hydrate: bool, previews: list[dict]) -> dict:
+    """The synthetic Home bucket: no folder => no repo/lane structure, one lane carries the rows."""
+    lane = {
+        "id": NO_PROJECT_ID, "label": NO_PROJECT_LABEL, "path": None, "isMain": False,
+        "isKanban": False, "sessions": homeless if hydrate else [],
+    }
+    home_repo = {
+        "id": NO_PROJECT_ID, "label": NO_PROJECT_LABEL, "path": None, "groups": [lane],
+        "sessionCount": len(homeless),
+    }
+    return _project_node(
+        NO_PROJECT_ID, NO_PROJECT_LABEL, None, [home_repo], len(homeless), _last_active(homeless),
+        previews, homeless, isNoProject=True)
 
 
 def build_tree(
@@ -494,40 +455,28 @@ def build_tree(
     hydrate: bool = False,
     is_junk_root: Optional[Callable[[str], bool]] = None,
     is_junk_cwd: Optional[Callable[[str], bool]] = None,
-    exists: Optional[Exists] = None,
-) -> dict:
-    """Build the authoritative project tree.
+    exists: Optional[Exists] = None) -> dict:
+    """Build the authoritative project tree -> ``{"projects", "scoped_session_ids"}``.
 
-    ``projects`` are ``projects_db.Project.to_dict()`` shapes (non-archived).
-    ``sessions`` are projected session-row dicts (``id``, ``cwd``, ``git_branch``,
-    ``git_repo_root``, ``started_at``, ``last_active``). ``discovered_repos`` are
-    ``{"root", "label", "sessions", "last_active"}``. ``is_junk_root`` flags git
-    roots that must never become an AUTO project (bare home dir, HERMES_HOME);
-    ``is_junk_cwd`` is the narrower policy for non-git session folders (selected
-    descendants may be intentional workspaces). User-created projects are honored
-    regardless. ``exists`` keeps a DELETED workspace (removed worktree, /tmp
-    scratch) from being promoted to a phantom AUTO project; omit it (remote
-    backends) to keep every candidate.
-
-    Returns ``{"projects": [...], "scoped_session_ids": [...]}``. With
-    ``hydrate`` False (overview) lane ``sessions`` are emptied but counts are
-    preserved and each project carries up to ``preview_limit`` ``previewSessions``;
-    True (drill-in) keeps full session rows.
+    ``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.
     """
     active_projects = [p for p in projects if not p.get("archived")]
     _junk = is_junk_root or (lambda _root: False)
     _junk_cwd = is_junk_cwd or (lambda _cwd: False)
     _exists = exists or (lambda _path: True)
     folder_index = _FolderIndex(active_projects)
-
-    by_project: dict[str, list[dict]] = {}
+    by_project: dict[str, list[dict]] = {}  # explicit project id -> owned rows
     unowned: list[dict] = []
     for session in sessions:
         owner = _project_for_session(session, folder_index, resolve)
-        if owner:
-            by_project.setdefault(owner["id"], []).append(session)
-        else:
-            unowned.append(session)
+        (by_project.setdefault(owner["id"], []) if owner else unowned).append(session)
 
     scoped_ids: list[str] = []
     result: list[dict] = []
@@ -544,90 +493,36 @@ def build_tree(
     for project in active_projects:
         psessions = by_project.get(project["id"], [])
         _scope(psessions)
-        result.append(
-            _project_node(
-                pid=project["id"],
-                label=project.get("name") or project["id"],
-                path=project.get("primary_path"),
-                color=project.get("color"),
-                icon=project.get("icon"),
-                repos=_seed_folder_repos(_build_repos(psessions, resolve, hydrate), project.get("folders") or [], resolve),
-                session_count=len(psessions),
-                last_active=_last_active(psessions),
-                preview_sessions=_previews(psessions),
-                sessions=psessions,
-            )
-        )
-
-    # Tier 2: auto projects from leftover sessions. Prefer the common git repo
-    # root, then fall back to the session cwd for historical/non-git workspaces
-    # (the pre-Projects desktop grouped every non-empty cwd; dropping that would
-    # flatten those sessions into Recents on upgrade).
-    by_auto_root: dict[str, dict] = {}
-    homeless: list[dict] = []  # every session no tier could place -> Home bucket
-
-    def _add_auto(root: str, session: dict) -> None:
-        key = _path_key(root)
-        if not key:
-            homeless.append(session)
-            return
-        by_auto_root.setdefault(key, {"root": root, "sessions": []})["sessions"].append(session)
-
-    for session in unowned:
-        root = _session_repo_root(session, resolve)
-        if root:
-            # A real git root uses the stricter repo policy; never reinterpret a
-            # filtered internal repo as a cwd-only project. A root no longer on
-            # disk is a stale persisted value and must not resurrect as a project.
-            if not _junk(root) and _exists(root):
-                _add_auto(root, session)
-            else:
-                homeless.append(session)
-            continue
-        cwd = (session.get("cwd") or "").strip()
-        if not cwd or _junk_cwd(cwd):
-            homeless.append(session)
-            continue
-        placement = _place_session(session, resolve)
-        # A placement that only echoes back an unresolvable cwd is the path-only
-        # heuristic guessing. If that dir is also gone from disk, promoting it
-        # mints a phantom project that can only be dismissed by hand -> Home.
-        if placement and _exists(placement["repo_key"]):
-            _add_auto(placement["repo_key"], session)
-        else:
-            homeless.append(session)
+        repos = _build_repos(psessions, resolve, hydrate)
+        repos = _seed_folder_repos(repos, project.get("folders") or [], resolve)
+        result.append(_project_node(
+            project["id"], project.get("name") or project["id"], project.get("primary_path"), repos,
+            len(psessions), _last_active(psessions), _previews(psessions), psessions,
+            color=project.get("color"), icon=project.get("icon")))
 
+    # Tier 2: auto projects from leftover sessions.
+    by_auto_root, homeless = _auto_buckets(unowned, resolve, _junk, _junk_cwd, _exists)
     seen: set[str] = set()
     for bucket in by_auto_root.values():
         auto_root, auto_sessions = bucket["root"], bucket["sessions"]
         auto_key = _path_key(auto_root)
         repos = _build_repos(auto_sessions, resolve, hydrate)
         repo_node = next(
-            (r for r in repos if _path_key(r.get("id") or r.get("path") or "") == auto_key), None
-        )
+            (r for r in repos if _path_key(r.get("id") or r.get("path") or "") == auto_key), None)
         if repo_node is None:
             homeless.extend(auto_sessions)
             continue
         seen.add(auto_key)
         _scope(auto_sessions)
-        result.append(
-            _project_node(
-                pid=auto_root,
-                label=base_name(auto_root) or auto_root,
-                path=auto_root,
-                repos=repos,
-                session_count=repo_node["sessionCount"],
-                last_active=_last_active(auto_sessions),
-                preview_sessions=_previews(auto_sessions),
-                sessions=auto_sessions,
-                is_auto=True,
-            )
-        )
+        result.append(_project_node(
+            auto_root, base_name(auto_root) or auto_root, auto_root, repos,
+            repo_node["sessionCount"], _last_active(auto_sessions), _previews(auto_sessions),
+            auto_sessions, isAuto=True))
 
     # Tier 3: repos discovered from full history / disk scan with no loaded
     # sessions, folded to their common root and not owned by an explicit project.
     for repo in discovered_repos or []:
-        raw_root = (repo.get("root") or "").strip()
+        raw_root = _field(repo, "root")
         if not raw_root:
             continue
         info = resolve(raw_root) if resolve else None
@@ -637,57 +532,19 @@ def build_tree(
             continue
         seen.add(root_key)
         label = repo.get("label") or base_name(root) or root
-        result.append(
-            _project_node(
-                pid=root,
-                label=label,
-                path=root,
-                repos=[_repo_node(root, label)],
-                session_count=int(repo.get("sessions") or 0),
-                last_active=float(repo.get("last_active") or 0),
-                preview_sessions=[],
-                is_auto=True,
-            )
-        )
+        result.append(_project_node(
+            root, label, root, [_repo_node(root, label)], int(repo.get("sessions") or 0),
+            float(repo.get("last_active") or 0), [], isAuto=True))
 
     # Auto projects are labelled by repo basename, which can collide; grow path
     # prefixes so each is distinct. Explicit projects keep their user-chosen names.
     _disambiguate_labels([p for p in result if p.get("isAuto")])
 
     # Tier 0: everything above could not place, so the grouped view loses no
-    # session. No folder => no repo/lane structure; the one synthetic lane just
-    # carries the rows. Leads the list; omitted entirely when empty.
+    # session. Leads the list; omitted entirely when empty.
     if homeless:
         homeless.sort(key=_session_time, reverse=True)
         _scope(homeless)
-        lane = {
-            "id": NO_PROJECT_ID,
-            "label": NO_PROJECT_LABEL,
-            "path": None,
-            "isMain": False,
-            "isKanban": False,
-            "sessions": homeless if hydrate else [],
-        }
-        home_repo = {
-            "id": NO_PROJECT_ID,
-            "label": NO_PROJECT_LABEL,
-            "path": None,
-            "groups": [lane],
-            "sessionCount": len(homeless),
-        }
-        result.insert(
-            0,
-            _project_node(
-                pid=NO_PROJECT_ID,
-                label=NO_PROJECT_LABEL,
-                path=None,
-                repos=[home_repo],
-                session_count=len(homeless),
-                last_active=_last_active(homeless),
-                preview_sessions=_previews(homeless),
-                sessions=homeless,
-                is_no_project=True,
-            ),
-        )
+        result.insert(0, _home_project(homeless, hydrate, _previews(homeless)))
 
     return {"projects": result, "scoped_session_ids": scoped_ids}