refactor(hermes_cli): uninstall/update_cmd_config/update_abort_recovery — _unlink_if, _ask_configure_new_options, safety-net line builders, _run_fresh_recovery_process/_parse_serve_units, compact docstrings

hermes_cli/uninstall.py

@@ -44,11 +44,9 @@ def _cancelled() -> None:
 def _confirm_yes(text: str) -> bool:
     """Ask the user to type ``yes``; False (after the cancel line, unless Ctrl-C/EOF) otherwise."""
     confirm = _prompt(f"Type '{color('yes', Colors.YELLOW)}' {text}: ")
-    if confirm == "yes":
-        return True
-    if confirm is not None:
+    if confirm != "yes" and confirm is not None:
         _cancelled()
-    return False
+    return confirm == "yes"
 
 
 def _remove_each(candidates, remove) -> list:
@@ -73,8 +71,7 @@ _SHELL_RC_NAMES = (".bashrc", ".bash_profile", ".profile", ".zshrc", ".zprofile"
 
 
 def _strip_hermes_path_lines(content: str) -> str:
-    """Drop the ``# Hermes Agent`` marker (+ its PATH line) and any hermes PATH line; squash blank
-    runs."""
+    """Drop the ``# Hermes Agent`` marker (+ its PATH line) and any hermes PATH line; squash blank runs."""
     new_lines = []
     skip_next = False
     for line in content.split('\n'):
@@ -96,18 +93,16 @@ def _strip_hermes_path_lines(content: str) -> str:
 
 def remove_path_from_shell_configs():
     """Remove Hermes PATH entries from shell configuration files."""
-    home = Path.home()
     removed_from = []
-    for config_path in (c for c in (home / n for n in _SHELL_RC_NAMES) if c.exists()):
+    for config_path in (c for c in (Path.home() / n for n in _SHELL_RC_NAMES) if c.exists()):
         try:
             content = config_path.read_text(encoding="utf-8")
             new_content = _strip_hermes_path_lines(content)
             if new_content != content:
                 from utils import atomic_write_text
                 # The user's own rc, never backed up: a bare write_text() truncates before the new
-                # content lands, and a crash mid-write (downgraded to a warning below) would leave
-                # an empty ~/.zshrc. Atomic replace also follows a symlinked rc (dotfiles repos);
-                # preserve_mode keeps its 0644 bits and owner (sudo-run uninstalls).
+                # content lands and a crash mid-write would leave an empty ~/.zshrc. Atomic replace
+                # also follows a symlinked rc; preserve_mode keeps its bits/owner (sudo-run uninstalls).
                 atomic_write_text(config_path, new_content, preserve_mode=True)
                 removed_from.append(config_path)
         except Exception as e:
@@ -120,10 +115,7 @@ def remove_wrapper_script():
     def _unlink_ours(wrapper: Path) -> bool:
         # Only our wrapper (contains a hermes_cli / hermes-agent reference).
         content = wrapper.read_text(encoding="utf-8")
-        if 'hermes_cli' not in content and 'hermes-agent' not in content:
-            return False
-        wrapper.unlink()
-        return True
+        return _unlink_if('hermes_cli' in content or 'hermes-agent' in content, wrapper)
 
     candidates = (
         bin_dir / name
@@ -132,6 +124,13 @@ def remove_wrapper_script():
     return _remove_each((w for w in candidates if w.exists()), _unlink_ours)
 
 
+def _unlink_if(ours: bool, path: Path) -> bool:
+    """Unlink *path* when it is ours; returns whether it was removed."""
+    if ours:
+        path.unlink()
+    return ours
+
+
 def _node_symlink_candidate_dirs() -> "list[Path]":
     """Directories where the installer may have placed node/npm/npx symlinks."""
     dirs: list[Path] = [Path.home() / ".local" / "bin"]
@@ -156,10 +155,7 @@ def remove_node_symlinks(hermes_home: Path) -> list:
         # os.readlink + manual join handles dangling links too (Path.resolve() on a dangling
         # link still returns the target path); the link must point into OUR node dir.
         target = (link.parent / os.readlink(link)).resolve()
-        if target != node_dir and node_dir not in target.parents:
-            return False
-        link.unlink()
-        return True
+        return _unlink_if(target == node_dir or node_dir in target.parents, link)
 
     candidates = (bin_dir / name for name in ("node", "npm", "npx") for bin_dir in _node_symlink_candidate_dirs())
     return _remove_each(candidates, _unlink_ours)
@@ -170,7 +166,6 @@ def uninstall_gateway_service():
     launchd / Scheduled Task + Startup folder). Termux/Android has neither: only the kill applies."""
     import platform
     stopped_something = False
-
     # 1. Kill any standalone gateway processes (all platforms, including Termux)
     try:
         from hermes_cli.gateway import kill_gateway_processes, find_gateway_pids
@@ -200,16 +195,14 @@ def _remove_systemd_gateway() -> bool:
     from hermes_cli.gateway import _systemctl_cmd, get_service_name, get_systemd_unit_path
     svc_name = get_service_name()
     removed_any = False
-
     for is_system, scope in ((False, "user"), (True, "system")):
         unit_path = get_systemd_unit_path(system=is_system)
         if not unit_path.exists():
             continue
         try:
-            if is_system and os.geteuid() != 0:  # windows-footgun: ok — Linux-only systemd path (dispatched on platform.system())
+            if is_system and os.geteuid() != 0:  # windows-footgun: ok — Linux-only systemd path
                 log_warn(f"System gateway service exists at {unit_path} but needs sudo to remove")
                 continue
-
             cmd = _systemctl_cmd(is_system)
             for verb in ("stop", "disable"):
                 subprocess.run(cmd + [verb, svc_name], capture_output=True, check=False)
@@ -237,17 +230,15 @@ def _remove_launchd_gateway() -> bool:
 def _remove_windows_gateway() -> bool:
     """Windows: uninstall Scheduled Task + Startup-folder entry via ``gateway_windows`` (it owns
     schtasks /Delete, the .cmd unlink and stopping the detached pythonw gateway)."""
-    from hermes_cli import gateway_windows
-    if not any(probe() for probe in (
-        gateway_windows.is_installed, gateway_windows.is_task_registered, gateway_windows.is_startup_entry_installed,
-    )):
+    from hermes_cli import gateway_windows as gw
+    if not any(probe() for probe in (gw.is_installed, gw.is_task_registered, gw.is_startup_entry_installed)):
         return False
     try:
-        gateway_windows.stop()
+        gw.stop()
     except Exception as e:
         log_warn(f"Could not stop Windows gateway cleanly: {e}")
     try:
-        gateway_windows.uninstall()
+        gw.uninstall()
         log_success("Removed Windows gateway (Scheduled Task + Startup entry)")
         return True
     except Exception as e:
@@ -261,29 +252,24 @@ _GATEWAY_SERVICE_REMOVERS = {
     "Darwin": (_remove_launchd_gateway, "Could not remove launchd gateway service"),
     "Windows": (_remove_windows_gateway, "Could not check Windows gateway service")}
 
-# Windows-specific helpers. install.ps1 does four things no rc file covers: (1) User-scope env
-# vars HERMES_HOME / HERMES_GIT_BASH_PATH in the registry (HKCU\Environment); (2) User-scope PATH
-# entries there too (%LOCALAPPDATA%\hermes\git\{cmd,bin,usr\bin}, ...\hermes\node); (3) PortableGit
-# + Node copies under %LOCALAPPDATA%\hermes\ (~200MB, useless after uninstall); (4) the
-# gateway-service dir (the Startup-folder entry is handled by gateway_windows.uninstall()).
-# Direct winreg writes are used instead of PowerShell one-liners: no subprocess, and they work
-# under Constrained Language Mode / restricted ExecutionPolicy. New shells see them immediately
-# (WM_SETTINGCHANGE would need ctypes and buys nothing — the user opens a new terminal anyway).
+# Windows helpers. install.ps1 leaves four things no rc file covers: User-scope env vars
+# HERMES_HOME / HERMES_GIT_BASH_PATH (HKCU\Environment), User-scope PATH entries
+# (%LOCALAPPDATA%\hermes\git\{cmd,bin,usr\bin}, ...\hermes\node), PortableGit + Node copies
+# (~200MB) and the gateway-service dir. Direct winreg writes (not PowerShell): no subprocess, and
+# they work under Constrained Language Mode; new shells see them without WM_SETTINGCHANGE.
 
 
 def _hermes_path_markers(hermes_home: Path, *, include_managed_bin: bool = False) -> list[str]:
-    """Prefixes identifying Hermes-owned User-PATH entries. ``include_managed_bin`` adds
-    ``<root>\bin`` (launchers + managed uv) — only when that dir is about to be deleted (full
-    uninstall from the default root), so a keep-data uninstall keeps the working uv resolvable."""
+    """Prefixes identifying Hermes-owned User-PATH entries (prefix match sweeps git\cmd, git\bin,
+    node...). ``include_managed_bin`` adds ``<root>\bin`` (launchers + managed uv) — only when that
+    dir is about to be deleted, so a keep-data uninstall keeps the working uv resolvable."""
     root = str(hermes_home).rstrip("\\/")
-    # Match on prefix so sub-entries (git\cmd, git\bin, git\usr\bin, node, etc.) all get swept.
     subs = ("hermes-agent", "git", "node", "venv") + (("bin",) if include_managed_bin else ())
     return [f"{root}\\{sub}" for sub in subs]
 
 
 def remove_path_from_windows_registry(hermes_home: Path, *, include_managed_bin: bool = False) -> list[str]:
-    """Strip Hermes-owned entries from User-scope PATH in the registry (see ``_hermes_path_markers``
-    for what ``include_managed_bin`` adds and when to pass it)."""
+    """Strip Hermes-owned entries from User-scope PATH in the registry (see ``_hermes_path_markers``)."""
     markers = tuple(m.lower() for m in _hermes_path_markers(hermes_home, include_managed_bin=include_managed_bin))
 
     def edit(winreg, key, removed):
@@ -304,7 +290,6 @@ def remove_path_from_windows_registry(hermes_home: Path, *, include_managed_bin:
 
 def remove_hermes_env_vars_windows() -> list[str]:
     """Delete HERMES_HOME and HERMES_GIT_BASH_PATH from User-scope env vars."""
-
     def edit(winreg, key, removed):
         for name in ("HERMES_HOME", "HERMES_GIT_BASH_PATH"):
             try:
@@ -339,9 +324,8 @@ def _edit_user_environment(edit, *, warn_label: str) -> list[str]:
 
 
 def remove_portable_tooling_windows(hermes_home: Path) -> list[Path]:
-    """Delete PortableGit and Node installs the Windows installer created under
-    ``%LOCALAPPDATA%\\hermes\\``.  Only called on full uninstall; they're
-    isolated from any system Git / Node so they cannot break other tools."""
+    """Delete the PortableGit / Node / gateway-service dirs the Windows installer created under
+    ``hermes_home`` (isolated from any system Git / Node, so nothing else breaks)."""
     targets = (hermes_home / sub for sub in ("git", "node", "gateway-service"))
     return _remove_each((t for t in targets if t.exists()), lambda t: shutil.rmtree(t) or True)
 
@@ -351,13 +335,10 @@ def remove_windows_bin_launchers(*, windows: bool | None = None) -> list[Path]:
     deletes the checkout, so a launcher pointing at ``<checkout>/venv`` would dangle (worse than
     command-not-found); the managed uv stays for keep-data reinstalls. Our own running trampoline is
     locked against deletion but not rename, so it is renamed aside with a non-executable suffix."""
-    if windows is None:
-        windows = _is_windows()
-    if not windows:
+    if not (_is_windows() if windows is None else windows):
         return []
     try:
-        # Lockstep launcher-name list — the same names install.ps1 and the
-        # startup heal stage into this dir.
+        # Lockstep launcher-name list — the same names install.ps1 and the startup heal stage here.
         from hermes_cli._install_repair import _WINDOWS_BIN_LAUNCHERS
         from hermes_constants import get_default_hermes_root
         bin_dir = get_default_hermes_root() / "bin"
@@ -390,9 +371,7 @@ def _is_default_hermes_home(hermes_home: Path) -> bool:
 
 
 def _discover_named_profiles():
-    """Return a list of ``ProfileInfo`` for every non-default profile, or ``[]`` if profile support is
-    unavailable or nothing is installed beyond the default root.
-    """
+    """``ProfileInfo`` for every non-default profile; ``[]`` when profile support is unavailable."""
     try:
         from hermes_cli.profiles import list_profiles
     except Exception:
@@ -431,7 +410,6 @@ def _uninstall_profile(profile) -> None:
             log_success(f"  Removed alias {alias_path}")
         except Exception as e:
             log_warn(f"  Could not remove alias {alias_path}: {e}")
-
     # 3. Wipe the profile's HERMES_HOME directory.
     _rmtree_step(profile.path, indent="  ", fully=False)
 
@@ -439,9 +417,7 @@ def _uninstall_profile(profile) -> None:
 def run_gui_uninstall(args):
     """``hermes uninstall --gui``: remove the desktop app's built artifacts, packaged bundle
     (best-effort) and Electron userData — never config/sessions/.env, the agent or its venv."""
-    from hermes_cli.gui_uninstall import (
-        agent_is_installed, gui_install_summary, uninstall_gui)
-
+    from hermes_cli.gui_uninstall import agent_is_installed, gui_install_summary, uninstall_gui
     hermes_home = get_hermes_home()
     summary = gui_install_summary(hermes_home)
     skip_confirm = bool(getattr(args, "yes", False))
@@ -608,8 +584,7 @@ def _print_uninstall_dry_run(*, project_root: Path, hermes_home: Path, full_unin
 
 
 def _remove_step(label: str, remove, success_fmt: str, none_msg: str) -> None:
-    """Announce ``label``, run ``remove()``, then log one success line per removed item (or
-    ``none_msg``)."""
+    """Announce ``label``, run ``remove()``, log one success line per removed item (or ``none_msg``)."""
     log_info(label)
     removed = remove()
     for item in removed:
@@ -643,18 +618,16 @@ def _perform_uninstall(
     print()
     print(color("Uninstalling...", Colors.CYAN, Colors.BOLD))
     print()
-
     # 1. Stop and uninstall gateway service + kill standalone processes
     log_info("Checking for running gateway...")
     if not uninstall_gateway_service():
         log_info("No gateway service or processes found")
 
-    # 2-3b. PATH entries (rc files, then the Windows User registry), wrapper, Windows launchers,
-    #    node symlinks. Windows: hermes_home is %VAR%-expanded because install.ps1 writes literal
-    #    C:\Users\<u>\AppData\Local\hermes\git\cmd; hermes\bin (launchers + managed uv) leaves the
-    #    PATH only when the full wipe below deletes it (keep-data keeps uv resolvable), while the
-    #    launchers themselves always go (see remove_windows_bin_launchers). Symlinks go only when
-    #    they still point into this home's node dir (never clobber nvm / user-managed Node).
+    # 2-3b. PATH entries, wrapper, Windows launchers, node symlinks. Windows: hermes_home is
+    #    %VAR%-expanded because install.ps1 writes literal C:\Users\<u>\...; hermes\bin (launchers +
+    #    managed uv) leaves the PATH only when the full wipe below deletes it (keep-data keeps uv
+    #    resolvable), while the launchers themselves always go. Symlinks go only when they still
+    #    point into this home's node dir (never clobber nvm / user-managed Node).
     windows = _is_windows()
     sweep_managed_bin = windows and full_uninstall and _is_default_hermes_home(hermes_home)
     for on_this_platform, label, remove, success_fmt, none_msg in (
@@ -675,9 +648,8 @@ def _perform_uninstall(
         if on_this_platform:
             _remove_step(label, remove, success_fmt, none_msg)
 
-    # 3c. Desktop Chat GUI artifacts: both flows remove the agent code, so the GUI goes with it.
-    #     uninstall_gui() never touches config/sessions/.env (safe in keep-data mode); the packaged
-    #     app + Electron userData live OUTSIDE HERMES_HOME, so the rmtree below wouldn't reach them.
+    # 3c. Chat GUI artifacts go with the agent code. uninstall_gui() never touches config/sessions/
+    #     .env (safe in keep-data mode); the packaged app + Electron userData live OUTSIDE HERMES_HOME.
     log_info("Removing desktop Chat GUI artifacts...")
     try:
         from hermes_cli.gui_uninstall import uninstall_gui
@@ -689,9 +661,8 @@ def _perform_uninstall(
     # 4. Remove installation directory (code) — we may be running from inside it.
     log_info("Removing installation directory...")
     _rmtree_step(project_root)
-
-    # 4b. Windows installer tooling under HERMES_HOME (PortableGit, Node, gateway-service) is not
-    #     user data: safe to remove in keep-data mode too.
+    # 4b. Windows installer tooling (PortableGit, Node, gateway-service) is not user data:
+    #     safe to remove in keep-data mode too.
     if windows:
         _remove_step(
             "Removing Windows installer artifacts (PortableGit, Node, gateway-service)...",
@@ -702,10 +673,8 @@ def _perform_uninstall(
     if full_uninstall:
         # 5a. Named profiles' homes live under <default>/profiles/ (swept by the rmtree below),
         #     but their services + alias scripts live OUTSIDE the default root.
-        if remove_profiles:
-            for prof in named_profiles:
-                _uninstall_profile(prof)
-
+        for prof in named_profiles if remove_profiles else ():
+            _uninstall_profile(prof)
         log_info("Removing configuration and data...")
         _rmtree_step(hermes_home)
     else:
@@ -753,8 +722,7 @@ class _UninstallArgs:
 
 def main(argv=None) -> int:
     """``python -m hermes_cli.uninstall --mode <gui|lite|full>``. Imports only stdlib +
-    ``hermes_constants`` + ``hermes_cli.colors`` (lazily ``gui_uninstall``), so it runs under a
-    bare system Python without the venv."""
+    ``hermes_constants`` + ``hermes_cli.colors``, so it runs under a bare system Python (no venv)."""
     import argparse
     parser = argparse.ArgumentParser(prog="python -m hermes_cli.uninstall")
     parser.add_argument(

hermes_cli/update_abort_recovery.py

@@ -1,7 +1,7 @@
 """Fresh-process recovery after the update's in-process restart phase aborts (the fleet restart
-runs in the interpreter that started before ``git pull``). Deliberately a separate owner from
-``update_cmd``: its own vocabulary (``verified`` / ``relaunch_attempted`` / ``failed``, serve units,
-survivors) and its own fail-closed contract."""
+runs in the interpreter that started before ``git pull``). Separate owner from ``update_cmd``: its
+own vocabulary (``verified`` / ``relaunch_attempted`` / ``failed``, serve units, survivors) and
+its own fail-closed contract."""
 
 from __future__ import annotations
 
@@ -36,12 +36,10 @@ def _surviving_pre_update_serve_runtimes(plan) -> list[dict]:
                 continue
             detail = getattr(runtime, "detail", None)
             planned[pid] = {
-                "pid": pid,
-                "kind": str(getattr(runtime, "kind", "")),
+                "pid": pid, "kind": str(getattr(runtime, "kind", "")),
                 "profile": str(getattr(runtime, "profile", "")),
                 "supervisor": str(getattr(runtime, "supervisor", "")),
-                "_create_time": _numeric(detail.get("create_time") if isinstance(detail, dict) else None),
-            }
+                "_create_time": _numeric(detail.get("create_time") if isinstance(detail, dict) else None)}
     except Exception as exc:
         logger.debug("Could not read planned serve runtimes: %s", exc)
         return []
@@ -56,20 +54,22 @@ def _surviving_pre_update_serve_runtimes(plan) -> list[dict]:
     except Exception as exc:
         logger.debug("Serve/dashboard survivor probe failed: %s", exc)
         live = None
-    survivors = []
-    for pid, row in planned.items():
-        if live is not None:
-            if pid not in live:
-                continue
-            planned_created, live_created = row["_create_time"], live[pid]
-            if (
-                planned_created is not None
-                and live_created is not None
-                and abs(float(live_created) - float(planned_created)) >= 2.0):
-                # Same number, different process: the pre-update runtime is gone
-                # and something new registered under its PID. Not a survivor.
-                continue
-        survivors.append(_without_incarnation(row))
+    def _still_live(pid, row) -> bool:
+        if live is None:
+            return True
+        if pid not in live:
+            return False
+        planned_created, live_created = row["_create_time"], live[pid]
+        # Same number, different process: the pre-update runtime is gone and something new
+        # registered under its PID. Not a survivor.
+        return not (
+            planned_created is not None and live_created is not None
+            and abs(float(live_created) - float(planned_created)) >= 2.0)
+
+    # The operator-facing row drops the incarnation (a matching key only).
+    survivors = [
+        {k: v for k, v in row.items() if k != "_create_time"}
+        for pid, row in planned.items() if _still_live(pid, row)]
     return sorted(survivors, key=lambda row: row["pid"])
 
 
@@ -77,11 +77,6 @@ def _numeric(value):
     return value if isinstance(value, (int, float)) else None
 
 
-def _without_incarnation(row: dict) -> dict:
-    """The operator-facing survivor row (incarnation is a matching key only)."""
-    return {key: value for key, value in row.items() if key != "_create_time"}
-
-
 def _qualified_serve_skips(skip_units) -> list[dict]:
     """Scope-qualify the units the aborted phase already settled: ``<scope>/<unit>`` because
     ``hermes-serve.service`` can exist in BOTH the user and system manager (two processes)."""
@@ -95,35 +90,11 @@ def _qualified_serve_skips(skip_units) -> list[dict]:
     return rows
 
 
-def _recover_gateway_restart_after_abort(
-    plan,
-    *,
-    gateway_mode: bool,
-    skip_profiles: set[str] | None = None,
-    skip_units: set[str] | None = None) -> dict[str, list]:
-    """Retry supervised gateway restarts from a clean Python process (the in-process restart ran
-    in the pre-``git pull`` interpreter). Only inventory-classified supervisor-owned profiles."""
-    from hermes_cli.update_cmd import _gateway_recovery_partition
-    candidates, skipped = _gateway_recovery_partition(plan, skip_profiles=skip_profiles)
-    profiles = sorted(candidates)
-    recover_serve = _serve_unit_recovery_available()
-    _empty_serve: dict[str, list] = {"verified": [], "failed": []}
-
-    def _result(requested, verified, relaunch_attempted, failed, serve_units=None) -> dict[str, list]:
-        return {
-            "requested": requested,
-            "verified": verified,
-            "relaunch_attempted": relaunch_attempted,
-            "failed": failed,
-            "skipped": skipped,
-            "serve_units": dict(_empty_serve) if serve_units is None else serve_units}
-
-    if not profiles and not recover_serve:
-        return _result([], [], [], [])
-
-    def _all_failed() -> dict[str, list]:
-        return _result(profiles, [], [], profiles)
-
+def _run_fresh_recovery_process(
+    profiles, candidates, *, gateway_mode: bool, recover_serve: bool, skip_units
+) -> "subprocess.CompletedProcess | None":
+    """Spawn ``hermes_cli.update_restart_recovery --stdin`` detached from this process; None when it
+    could not run (no systemd-run in gateway mode, OSError, timeout) — the caller fails closed."""
     command = [sys.executable, "-m", "hermes_cli.update_restart_recovery", "--stdin"]
     env = os.environ.copy()
     env["HERMES_UPDATE_RESTART_RECOVERY"] = "1"
@@ -137,66 +108,83 @@ def _recover_gateway_restart_after_abort(
         systemd_run = shutil.which("systemd-run")
         if not systemd_run:
             logger.warning("Cannot isolate fresh gateway recovery from the gateway cgroup")
-            return _all_failed()
+            return None
         command = [systemd_run, "--user", "--scope", "--quiet", "--collect", "--", *command]
 
     kwargs = {
-        "input": json.dumps(
-            {
-                "profiles": profiles,
-                "supervisors": candidates,
-                "serve_units": {
-                    "recover": recover_serve, "skip": _qualified_serve_skips(skip_units)}}),
-        "capture_output": True,
-        "text": True,
-        "encoding": "utf-8",
-        "errors": "replace",
-        "check": False,
-        "env": env,
+        "input": json.dumps({
+            "profiles": profiles, "supervisors": candidates,
+            "serve_units": {"recover": recover_serve, "skip": _qualified_serve_skips(skip_units)}}),
+        "capture_output": True, "text": True, "encoding": "utf-8", "errors": "replace",
+        "check": False, "env": env,
         # Gateway profiles run sequentially at up to 90s each, plus the serve pass's own
         # restart + settle budget — don't kill a recovery that was working.
         "timeout": max(180, 30 + 90 * len(profiles) + (150 if recover_serve else 0))}
     if sys.platform == "win32":
         kwargs["creationflags"] = (
-            getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
-            | getattr(subprocess, "DETACHED_PROCESS", 0))
+            getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr(subprocess, "DETACHED_PROCESS", 0))
     else:
         kwargs["start_new_session"] = True
-
     try:
-        result = subprocess.run(command, **kwargs)
+        return subprocess.run(command, **kwargs)
     except (OSError, subprocess.TimeoutExpired) as exc:
         logger.warning("Fresh gateway restart recovery failed: %s", exc)
-        return _all_failed()
+        return None
 
+
+def _parse_serve_units(raw_serve, *, recover_serve: bool) -> dict[str, list]:
+    """Validate the child's ``serve_units`` block; an unreadable block is not "nothing to do" when
+    serve recovery was requested (those units may still serve the pre-update generation)."""
+    if (
+        isinstance(raw_serve, dict)
+        and isinstance(raw_serve.get("verified"), list) and isinstance(raw_serve.get("failed"), list)
+        and all(isinstance(unit, str) for unit in (*raw_serve["verified"], *raw_serve["failed"]))):
+        return {"verified": sorted(raw_serve["verified"]), "failed": sorted(raw_serve["failed"])}
+    if recover_serve:
+        logger.warning("Fresh recovery returned an invalid serve-unit result")
+        return {"verified": [], "failed": ["<unreadable>"]}
+    return {"verified": [], "failed": []}
+
+
+def _recover_gateway_restart_after_abort(
+    plan, *, gateway_mode: bool, skip_profiles: set[str] | None = None,
+    skip_units: set[str] | None = None) -> dict[str, list]:
+    """Retry supervised gateway restarts from a clean Python process (the in-process restart ran
+    in the pre-``git pull`` interpreter). Only inventory-classified supervisor-owned profiles."""
+    from hermes_cli.update_cmd import _gateway_recovery_partition
+    candidates, skipped = _gateway_recovery_partition(plan, skip_profiles=skip_profiles)
+    profiles = sorted(candidates)
+    recover_serve = _serve_unit_recovery_available()
+
+    def _result(requested, verified, relaunch_attempted, failed, serve_units=None) -> dict[str, list]:
+        return {
+            "requested": requested, "verified": verified, "relaunch_attempted": relaunch_attempted,
+            "failed": failed, "skipped": skipped,
+            "serve_units": {"verified": [], "failed": []} if serve_units is None else serve_units}
+
+    if not profiles and not recover_serve:
+        return _result([], [], [], [])
+
+    def _all_failed() -> dict[str, list]:
+        return _result(profiles, [], [], profiles)
+
+    result = _run_fresh_recovery_process(
+        profiles, candidates, gateway_mode=gateway_mode, recover_serve=recover_serve, skip_units=skip_units)
+    if result is None:
+        return _all_failed()
     if result.returncode != 0:
         logger.warning("Fresh gateway restart recovery exited %s", result.returncode)
         return _all_failed()
-
     try:
         recovery_result = json.loads(result.stdout or "")
         verified = recovery_result.get("verified")
         relaunch_attempted = recovery_result.get("relaunch_attempted")
         failed = recovery_result.get("failed")
-        raw_serve = recovery_result.get("serve_units") or dict(_empty_serve)
+        raw_serve = recovery_result.get("serve_units") or {"verified": [], "failed": []}
     except (AttributeError, TypeError, ValueError):
         logger.warning("Fresh gateway restart recovery returned invalid JSON")
         return _all_failed()
-
-    serve_units = dict(_empty_serve)
-    if (
-        isinstance(raw_serve, dict)
-        and isinstance(raw_serve.get("verified"), list)
-        and isinstance(raw_serve.get("failed"), list)
-        and all(
-            isinstance(unit, str) for unit in (*raw_serve["verified"], *raw_serve["failed"]))):
-        serve_units = {
-            "verified": sorted(raw_serve["verified"]), "failed": sorted(raw_serve["failed"])}
-    elif recover_serve:
-        # An unreadable serve block is not "nothing to do": those units host tui_gateway and
-        # may still be serving the pre-update generation.
-        logger.warning("Fresh recovery returned an invalid serve-unit result")
-        serve_units = {"verified": [], "failed": ["<unreadable>"]}
+    serve_units = _parse_serve_units(raw_serve, recover_serve=recover_serve)
 
     buckets = (verified, relaunch_attempted, failed)
     reported: list[str] = []
@@ -205,8 +193,7 @@ def _recover_gateway_restart_after_abort(
     if (
         not all(isinstance(bucket, list) for bucket in buckets)
         or any(not isinstance(profile, str) for profile in reported)
-        or set(reported) != set(profiles)
-        or len(reported) != len(set(reported))):
+        or set(reported) != set(profiles) or len(reported) != len(set(reported))):
         logger.warning("Fresh gateway restart recovery returned incomplete profiles")
         return _all_failed()
 
@@ -225,21 +212,18 @@ def _recover_gateway_restart_after_abort(
 def _warn_stale_serve_runtimes(rows) -> None:
     """Name the serve/dashboard processes still on pre-update code: ``hermes serve`` hosts
     ``tui_gateway.server``, and an un-restarted unit keeps the pre-pull ``sys.modules`` graph so
-    every chat turn fails with an ``ImportError`` no gateway row explains. Print PIDs + the fixing
-    command."""
+    every chat turn fails with an ``ImportError`` no gateway row explains."""
     if not rows:
         return
     print(
         "  ⚠ These serve/dashboard processes still run pre-update code"
         " (they started before the checkout changed):")
     for row in rows:
-        supervisor = row.get("supervisor") or "unknown"
         print(
             f"      pid {row.get('pid')} — {row.get('kind')}"
-            f" (profile {row.get('profile') or 'default'}, {supervisor})")
+            f" (profile {row.get('profile') or 'default'}, {row.get('supervisor') or 'unknown'})")
     print(
-        "    Restart them before using Hermes again, e.g."
-        " `systemctl --user restart hermes-serve.service`"
+        "    Restart them before using Hermes again, e.g. `systemctl --user restart hermes-serve.service`"
         " or by relaunching `hermes serve` / the Desktop app.")
 
 
@@ -250,13 +234,10 @@ def _abort_recovery_is_complete(
     family is accounted for. Empty ``planned_gateway_profiles`` is deliberately NOT completeness:
     with no gateway leg to prove, ``_restart_phase_failure_is_incomplete`` + the stale rows decide.
     """
-    if not planned_gateway_profiles:
-        return False
-    if not set(planned_gateway_profiles) <= set(covered_gateway_profiles):
-        return False
     result = recovery_result or {}
-    if result.get("failed") or result.get("relaunch_attempted"):
-        return False
-    if (result.get("serve_units") or {}).get("failed"):
-        return False
-    return not stale_runtime_rows
+    return bool(
+        planned_gateway_profiles
+        and set(planned_gateway_profiles) <= set(covered_gateway_profiles)
+        and not (result.get("failed") or result.get("relaunch_attempted"))
+        and not (result.get("serve_units") or {}).get("failed")
+        and not stale_runtime_rows)

hermes_cli/update_cmd_config.py

@@ -20,11 +20,8 @@ def _reload_config_modules() -> None:
     import importlib
     importlib.invalidate_caches()
     for mod_name in (
-        "hermes_cli.config_defaults",
-        "hermes_cli.config",
-        "hermes_cli.config_migrations",
-        "hermes_cli._subprocess_compat",
-        "hermes_cli.dashboard_procs"):
+        "hermes_cli.config_defaults", "hermes_cli.config", "hermes_cli.config_migrations",
+        "hermes_cli._subprocess_compat", "hermes_cli.dashboard_procs"):
         mod = sys.modules.get(mod_name)
         if mod is not None:
             try:
@@ -34,8 +31,7 @@ def _reload_config_modules() -> None:
 
 
 def _run_config_check_fresh() -> tuple:
-    """Return ``(current_ver, latest_ver)`` using freshly-reloaded modules (see
-    ``_reload_config_modules``)."""
+    """``(current_ver, latest_ver)`` from freshly-reloaded modules (see ``_reload_config_modules``)."""
     from hermes_cli.update_cmd import _reload_config_modules
     _reload_config_modules()
     from hermes_cli.config import check_config_version
@@ -43,8 +39,7 @@ def _run_config_check_fresh() -> tuple:
 
 
 def _run_migrate_config_fresh(*, interactive: bool = False, quiet: bool = False) -> dict:
-    """Run config migration using freshly-reloaded modules (see ``_reload_config_modules``); returns
-    results dict."""
+    """Run config migration with freshly-reloaded modules; returns the results dict."""
     from hermes_cli.update_cmd import _reload_config_modules
     _reload_config_modules()
     from hermes_cli.config import migrate_config
@@ -52,12 +47,10 @@ def _run_migrate_config_fresh(*, interactive: bool = False, quiet: bool = False)
 
 
 def _migrate_sibling_profile_configs() -> list[tuple[str, int, int]]:
-    """Migrate every SIBLING profile's config.yaml (the shared checkout serves all profiles;
-    siblings were left on configs the new code couldn't read). Per sibling (active skipped): scope
-    via the context-local HERMES_HOME override (never ``os.environ``) and run the NON-INTERACTIVE
-    quiet migration — prompt-requiring settings wait for that profile's own session. Returns
-    ``[(name, from_version, to_version), ...]``; never raises (a failing profile falls back to its
-    startup migration)."""
+    """Migrate every SIBLING profile's config.yaml (the shared checkout serves all profiles). Per
+    sibling (active skipped): scope via the context-local HERMES_HOME override (never ``os.environ``)
+    and run the NON-INTERACTIVE quiet migration — prompt-requiring settings wait for that profile's
+    own session. Returns ``[(name, from_version, to_version), ...]``; never raises."""
     from hermes_cli.update_cmd import _run_config_check_fresh, _run_migrate_config_fresh
     migrated: list[tuple[str, int, int]] = []
     with _best_effort('Sibling profile enumeration failed: %s'):
@@ -95,86 +88,82 @@ def _migrate_sibling_profile_configs() -> list[tuple[str, int, int]]:
 
 
 def _restore_snapshot_safety_nets(pre_update_snapshot_id) -> None:
-    """Post-migration safety nets: restore cron jobs / protected model settings lost during the update,
-    for the active profile (from *pre_update_snapshot_id*) and every sibling profile (own snapshot)."""
-    # Safety net: migrations/desktop scheduler have emptied or truncated cron/jobs.json;
-    # restore from the pre-update snapshot if jobs went missing.
-    try:
+    """Post-migration safety nets (never break an otherwise-good update): restore cron jobs
+    emptied by migrations/the desktop scheduler and protected model settings (model.provider /
+    model.default / moa:) rewritten by Desktop repair cycles — for the active profile (from
+    *pre_update_snapshot_id*) and every sibling profile (against ITS OWN snapshot)."""
+    def _cron_line(r):
+        return (
+            f"cron/jobs.json lost jobs during this update — restored {r['job_count']} job(s) "
+            f"from pre-update snapshot {r['snapshot_id']}.")
+
+    def _cfg_line(r):
+        return (
+            f"config.yaml user model settings were rewritten during this update — restored "
+            f"{', '.join(r['keys'])} from pre-update snapshot {r['snapshot_id']}.")
+
+    with _best_effort("Cron jobs auto-restore check failed: %s"):
         from hermes_cli.backup import restore_cron_jobs_if_emptied
         cron_restore = restore_cron_jobs_if_emptied(pre_update_snapshot_id)
         if cron_restore:
             print()
-            print(
-                "  ⚠️  cron/jobs.json lost jobs during this update — "
-                f"restored {cron_restore['job_count']} job(s) from "
-                f"pre-update snapshot {cron_restore['snapshot_id']}.")
-    except Exception as exc:
-        # Never let the cron safety net break an otherwise-good update.
-        logger.debug("Cron jobs auto-restore check failed: %s", exc)
-
-    # Desktop update/repair cycles have rewritten model.provider/model.default and dropped
-    # moa:; restore only those protected keys from the same pre-update snapshot.
-    try:
+            print(f"  ⚠️  {_cron_line(cron_restore)}")
+    with _best_effort("Config model-settings auto-restore check failed: %s"):
         from hermes_cli.backup import restore_config_model_settings_if_rewritten
         cfg_restore = restore_config_model_settings_if_rewritten(pre_update_snapshot_id)
         if cfg_restore:
             print()
-            print(
-                "  ⚠️  config.yaml user model settings were rewritten during "
-                f"this update — restored {', '.join(cfg_restore['keys'])} "
-                f"from pre-update snapshot {cfg_restore['snapshot_id']}.")
-    except Exception as exc:
-        # Never let the config safety net break an otherwise-good update.
-        logger.debug("Config model-settings auto-restore check failed: %s", exc)
-
-    # Same cron-jobs safety net per sibling profile against ITS OWN pre-update snapshot.
+            print(f"  ⚠️  {_cfg_line(cfg_restore)}")
     with _best_effort('Sibling cron auto-restore check failed: %s'):
         from hermes_cli.backup import restore_cron_jobs_all_profiles
-        for _restored in restore_cron_jobs_all_profiles(
-            _LAST_SIBLING_SNAPSHOTS):
+        for _restored in restore_cron_jobs_all_profiles(_LAST_SIBLING_SNAPSHOTS):
             print()
-            print(
-                f"  ⚠️  Profile '{_restored['profile']}': cron/jobs.json "
-                f"lost jobs during this update — restored "
-                f"{_restored['job_count']} job(s) from pre-update "
-                f"snapshot {_restored['snapshot_id']}.")
-
-    # Same config model-settings safety net for sibling profiles.
+            print(f"  ⚠️  Profile '{_restored['profile']}': {_cron_line(_restored)}")
     with _best_effort('Sibling config auto-restore check failed: %s'):
         from hermes_cli.backup import restore_config_model_settings_all_profiles
-        for _cfg_restored in restore_config_model_settings_all_profiles(
-            _LAST_SIBLING_SNAPSHOTS):
+        for _cfg_restored in restore_config_model_settings_all_profiles(_LAST_SIBLING_SNAPSHOTS):
             print()
-            print(
-                f"  ⚠️  Profile '{_cfg_restored['profile']}': config.yaml "
-                f"user model settings were rewritten during this update — "
-                f"restored {', '.join(_cfg_restored['keys'])} from "
-                f"pre-update snapshot {_cfg_restored['snapshot_id']}.")
+            print(f"  ⚠️  Profile '{_cfg_restored['profile']}': {_cfg_line(_cfg_restored)}")
+
+
+def _ask_configure_new_options(*, assume_yes: bool, gateway_mode: bool) -> str:
+    """The yes/no answer for "configure new options now?": "y" under --yes, the messenger's reply in
+    gateway mode, "auto" (safe migrations only) when non-interactive, else ``input()``."""
+    from hermes_cli.update_cmd import _gateway_prompt
+    if assume_yes:
+        print("  ℹ --yes: auto-applying config migration (skipping API-key prompts).")
+        return "y"
+    if gateway_mode:
+        return _gateway_prompt("Would you like to configure new options now? [Y/n]", "n").strip().lower()
+    if not (sys.stdin.isatty() and sys.stdout.isatty()):
+        print("  ℹ Non-interactive session — applying safe config migrations.")
+        return "auto"
+    try:
+        return input("Would you like to configure them now? [Y/n]: ").strip().lower()
+    except EOFError:
+        return "n"
+    except UnicodeDecodeError:
+        # Non-UTF-8 locales / embedded terminals can make input() raise this.
+        print(
+            "  ⚠ Could not read input (encoding issue). Skipping. "
+            "Run 'hermes config migrate' manually to configure.")
+        return "n"
 
 
 def _check_and_apply_config_migration(
-    *,
-    assume_yes: bool = False,
-    gateway_mode: bool = False,
-    pre_update_snapshot_id: str | None = None) -> None:
+    *, assume_yes: bool = False, gateway_mode: bool = False, pre_update_snapshot_id: str | None = None
+) -> None:
     """Check/apply config migrations with freshly-reloaded modules. Runs on EVERY completion path
     (post-pull, venv-repair, Node-deps repair on ``commit_count == 0``) so an interrupted update
     that already pulled code doesn't strand an old config version."""
     from hermes_cli.update_cmd import (
-        _gateway_prompt,
-        _migrate_sibling_profile_configs,
-        _reload_config_modules,
-        _run_config_check_fresh,
+        _migrate_sibling_profile_configs, _reload_config_modules, _run_config_check_fresh,
         _run_migrate_config_fresh)
     print()
     print("→ Checking configuration for new options...")
-
     # Reload BEFORE any config reads so all checks use the updated code.
     _reload_config_modules()
-
-    from hermes_cli.config import (
-        get_missing_env_vars, get_missing_config_fields)
-
+    from hermes_cli.config import get_missing_env_vars, get_missing_config_fields
     # A config-check failure must not break an otherwise-successful update.
     try:
         missing_env = get_missing_env_vars(required_only=True)
@@ -218,41 +207,17 @@ def _check_and_apply_config_migration(
             _print_items(missing_config, "New options", "key")
 
         print()
-        if assume_yes:
-            print("  ℹ --yes: auto-applying config migration (skipping API-key prompts).")
-            response = "y"
-        elif gateway_mode:
-            response = (
-                _gateway_prompt(
-                    "Would you like to configure new options now? [Y/n]", "n")
-                .strip()
-                .lower())
-        elif not (sys.stdin.isatty() and sys.stdout.isatty()):
-            print("  ℹ Non-interactive session — applying safe config migrations.")
-            response = "auto"
-        else:
-            try:
-                response = (input("Would you like to configure them now? [Y/n]: ").strip().lower())
-            except EOFError:
-                response = "n"
-            except UnicodeDecodeError:
-                # Non-UTF-8 locales / embedded terminals can make input() raise this.
-                print(
-                    "  ⚠ Could not read input (encoding issue). Skipping. "
-                    "Run 'hermes config migrate' manually to configure.")
-                response = "n"
-
+        response = _ask_configure_new_options(assume_yes=assume_yes, gateway_mode=gateway_mode)
         if response in {"", "y", "yes", "auto"}:
             print()
             # Gateway/--yes/non-interactive can't prompt for API keys; still run the
             # non-interactive pass so defaults and version bumps land before the gateway restarts.
-            interactive_migration = not (gateway_mode or assume_yes or response == "auto")
-            results = _run_migrate_config_fresh(interactive=interactive_migration, quiet=False)
-
+            unattended = gateway_mode or assume_yes or response == "auto"
+            results = _run_migrate_config_fresh(interactive=not unattended, quiet=False)
             if results["env_added"] or results["config_added"]:
                 print()
                 print("✓ Configuration updated!")
-            if (gateway_mode or assume_yes or response == "auto") and missing_env:
+            if unattended and missing_env:
                 print("  ℹ API keys require manual entry: hermes config migrate")
         else:
             print()
@@ -263,9 +228,8 @@ def _check_and_apply_config_migration(
     # The migration above touched only the active profile; run the same NON-INTERACTIVE
     # migration per sibling home via the context-local HERMES_HOME override (never os.environ).
     with _best_effort('Sibling config migration failed: %s'):
-        _migrated_siblings = _migrate_sibling_profile_configs()
-        for _name, _from_ver, _to_ver in _migrated_siblings:
-            print(f"  ✓ Profile '{_name}': config format updated " f"(v{_from_ver} → v{_to_ver})")
+        for _name, _from_ver, _to_ver in _migrate_sibling_profile_configs():
+            print(f"  ✓ Profile '{_name}': config format updated (v{_from_ver} → v{_to_ver})")
 
     _restore_snapshot_safety_nets(pre_update_snapshot_id)
 
@@ -281,17 +245,12 @@ def _print_items(items, label, key, fallback_key=None):
     print(f"  {label}:")
     shown = items[:8]
     for it in shown:
+        # Defensive: some callers/mocks pass bare name strings.
         if isinstance(it, dict):
             name = it.get(key) or (fallback_key and it.get(fallback_key)) or "?"
             desc = (it.get("description") or "").strip()
         else:
-            # Defensive: some callers/mocks pass bare name strings.
-            name = str(it)
-            desc = ""
-        if desc:
-            print(f"      • {name} — {desc}")
-        else:
-            print(f"      • {name}")
-    extra = len(items) - len(shown)
-    if extra > 0:
-        print(f"      … and {extra} more")
+            name, desc = str(it), ""
+        print(f"      • {name} — {desc}" if desc else f"      • {name}")
+    if len(items) > len(shown):
+        print(f"      … and {len(items) - len(shown)} more")