From b5605b7a8219e15dee52c4e1636e7f2a319573c9 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:52:18 -0700
Subject: [PATCH 1/4] =?UTF-8?q?refactor(hermes=5Fcli/update):=20compact=20?=
 =?UTF-8?q?Windows=20gateway=20pause/resume=20module=20=E2=80=94=20walrus?=
 =?UTF-8?q?=20rungs,=20poll=20helper=20reuse,=20shared=20purpose=20tuple?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 hermes_cli/update_cmd_windows.py | 365 ++++++++++++-------------------
 1 file changed, 139 insertions(+), 226 deletions(-)

diff --git a/hermes_cli/update_cmd_windows.py b/hermes_cli/update_cmd_windows.py
index 6dc9585ff0..76def34f88 100644
--- a/hermes_cli/update_cmd_windows.py
+++ b/hermes_cli/update_cmd_windows.py
@@ -17,8 +17,9 @@ from pathlib import Path
 
 from hermes_cli.update_cmd_common import _best_effort
 
-# Log-record parity with the origin module.
-logger = logging.getLogger("hermes_cli.update_cmd")
+logger = logging.getLogger("hermes_cli.update_cmd")  # log-record parity with the origin module
+
+_BACKEND_PURPOSES = ("serve", "dashboard")
 
 
 @contextmanager
@@ -35,11 +36,8 @@ def _write_update_planned_stop_marker(profile_path: Path, pid: int) -> bool:
     try:
         from gateway.status import _get_process_start_time
         from utils import atomic_json_write
-
-        record = {
-            "target_pid": pid, "target_start_time": _get_process_start_time(pid),
-            "stopper_pid": os.getpid(), "written_at": datetime.now(timezone.utc).isoformat(),
-        }
+        record = {"target_pid": pid, "target_start_time": _get_process_start_time(pid),
+                  "stopper_pid": os.getpid(), "written_at": datetime.now(timezone.utc).isoformat()}
         atomic_json_write(Path(profile_path) / ".gateway-planned-stop.json", record, indent=None, separators=(",", ":"))
         return True
     except (OSError, PermissionError):
@@ -50,7 +48,6 @@ def _wait_for_windows_update_gateway_exit(pids: list[int], *, timeout: float) ->
     """Wait for the given gateway PIDs to exit, returning survivors."""
     if not pids:
         return set()
-
     from gateway.status import _pid_exists
 
     def _alive(pid: int) -> bool:
@@ -72,8 +69,8 @@ def _self_and_non_gateway_ancestor_pids(psutil) -> set[int]:
     """PIDs a venv-holder scan must never nominate: this process and its non-gateway ancestry.
 
     Do NOT blanket-exclude ancestors: under ``/update`` the updater is a CHILD of the gateway, and hiding it
-    dead-ends the update on ``venv-blocked``. Keep GATEWAY ancestors visible (the pause path stops them
-    gracefully; a detached child survives on Windows); never nominate interactive ancestry as a blocker.
+    dead-ends the update on ``venv-blocked``. Gateway ancestors stay visible (the pause path stops them
+    gracefully; a detached child survives on Windows); interactive ancestry is never a blocker.
     """
     try:
         from gateway.status import looks_like_gateway_command_line as _is_gw
@@ -125,21 +122,19 @@ def _detect_venv_python_processes(*, exclude_pids: set[int] | None = None) -> li
 
     The hermes.exe shim guard misses the Desktop backend and anything off ``venv\\Scripts\\python(w).exe``;
     they keep ``.pyd`` files mapped so a mid-update dependency sync dies half-way. Empty off-Windows / without
-    psutil; self + non-gateway ancestors excluded.
+    psutil; self + non-gateway ancestors excluded. cmdline/cwd are expensive per process on Windows (500+
+    procs can blow the Desktop preflight watchdog), so they are fetched lazily for plausible candidates only.
+    The FULL cmdline is kept: callers parse it (the pausable-gateway exemption looks for ``gateway run``).
     """
     from hermes_cli.update_cmd import _m
     psutil = _psutil()
     if not _m()._is_windows() or psutil is None:
         return []
-
     venv_prefix = _lower_dir_prefix(_m().PROJECT_ROOT / "venv")
     root_prefix = _lower_dir_prefix(_m().PROJECT_ROOT)
     skip = set(exclude_pids or set()) | _self_and_non_gateway_ancestor_pids(psutil)
-
     matches: list[tuple[int, str, str]] = []
     try:
-        # cmdline/cwd are expensive per-process on Windows (500+ procs can blow the
-        # Desktop preflight watchdog): fetch them lazily for plausible candidates only.
         proc_iter = psutil.process_iter(["pid", "exe", "name"])
     except Exception:
         return []
@@ -174,8 +169,6 @@ def _detect_venv_python_processes(*, exclude_pids: set[int] | None = None) -> li
                 cwd_low = os.sep
             is_holder = root_prefix in cmdline_low or cwd_low.startswith(root_prefix)
         if is_holder:
-            # FULL cmdline: callers parse it (pausable-gateway exemption looks for `gateway run`);
-            # truncating here misreported autostarted gateways as blockers. Truncate at display time.
             matches.append((int(pid), name, cmdline_raw))
     return matches
 
@@ -200,9 +193,7 @@ def _holder_value_flags() -> frozenset:
     flags: set[str] = {"--profile", "-p", "--config"}
     try:
         from hermes_cli._parser import build_top_level_parser
-
-        parser = build_top_level_parser()[0]
-        for action in parser._actions:
+        for action in build_top_level_parser()[0]._actions:
             if action.option_strings and action.nargs != 0:
                 flags.update(action.option_strings)
         _holder_value_flags_cache = frozenset(flags)
@@ -231,7 +222,6 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None:
     entry_idx = next((i for i, token in enumerate(tokens) if _is_entry(i, token)), None)
     if entry_idx is None:
         return None
-
     value_flags = _holder_value_flags()
     i = entry_idx + 1
     while i < len(tokens):
@@ -251,12 +241,12 @@ def _format_venv_python_holders_message(matches: list[tuple[int, str, str]]) ->
     Labels come from the parsed SUBCOMMAND, never substring: a standalone ``hermes dashboard`` must not be
     called the Desktop backend, ``--preserve-cache`` must not match "serve". Unknown argv gets no hint.
     """
-    lines = ["✗ Other Hermes processes are running from this install's venv:"]
     hint_by_subcommand = {
         "serve": "  ← Hermes backend (if the Desktop app is open, close it)",
         "dashboard": "  ← hermes dashboard (stop it: hermes dashboard stop, or close that terminal)",
         "gateway": "  ← gateway",
     }
+    lines = ["✗ Other Hermes processes are running from this install's venv:"]
     for pid, name, cmdline in matches[:6]:
         hint = hint_by_subcommand.get(_hermes_holder_subcommand(cmdline) or "", "")
         lines.append(f"  PID {pid}  {name}  {cmdline[:120]}{hint}")
@@ -284,9 +274,8 @@ def _venv_launcher_ancestors(pids: list[int]) -> list[int]:
     psutil = _psutil()
     if not _m()._is_windows() or not pids or psutil is None:
         return []
-
     venv_prefix = _lower_dir_prefix(_m().PROJECT_ROOT / "venv")
-    skip = _self_and_non_gateway_ancestor_pids(psutil)
+    skip = _self_and_non_gateway_ancestor_pids(psutil) | set(pids)
     found: list[int] = []
     for pid in pids:
         with suppress(Exception):
@@ -294,9 +283,7 @@ def _venv_launcher_ancestors(pids: list[int]) -> list[int]:
             if parent is None:
                 continue
             ppid = int(parent.pid)
-            if ppid in skip or ppid in found or ppid in set(pids):
-                continue
-            if (parent.exe() or "").lower().startswith(venv_prefix):
+            if ppid not in skip and ppid not in found and (parent.exe() or "").lower().startswith(venv_prefix):
                 found.append(ppid)
     return found
 
@@ -309,7 +296,6 @@ def _leftover_pausable_gateway_pids(matches: list[tuple[int, str, str]]) -> list
     re-read via psutil when possible since the scan may hold only a cmdline prefix.
     """
     from hermes_cli._scan_venv_blockers import _is_pausable_gateway
-
     psutil = _psutil()
     pids: list[int] = []
     for pid, _name, cmdline in matches:
@@ -332,17 +318,15 @@ def _refuse_gateway_ancestor_tree_kill(pids: list[int], *, gateway_mode: bool) -
         return False
     try:
         from hermes_cli.gateway import _is_pid_ancestor_of_current_process
-
         ancestors = [int(pid) for pid in pids if _is_pid_ancestor_of_current_process(int(pid))]
     except Exception as exc:
         logger.debug("Could not inspect gateway ancestry before tree-kill: %s", exc)
         return False
     if not ancestors:
         return False
-    rendered = ", ".join(str(pid) for pid in ancestors)
     print(
         "✗ Refusing to stop the gateway process tree because this updater "
-        f"is running inside it (gateway PID(s): {rendered}).\n"
+        f"is running inside it (gateway PID(s): {', '.join(str(pid) for pid in ancestors)}).\n"
         "  On Windows, taskkill /T would terminate the updater before the update can run.\n"
         "  From a chat platform, use `/update` instead.\n"
         "  Otherwise, run `hermes update` from a separate terminal."
@@ -363,11 +347,8 @@ def _ledger_manual_serve_holders(matches: list[tuple[int, str, str]]) -> list[di
         return []
     holder_pids = {int(pid) for pid, _name, _cmd in matches}
     return [
-        entry
-        for entry in ledger_entries()
-        if entry.get("purpose") in ("serve", "dashboard")
-        and isinstance(entry.get("pid"), int)
-        and entry["pid"] in holder_pids
+        entry for entry in ledger_entries()
+        if entry.get("purpose") in _BACKEND_PURPOSES and isinstance(entry.get("pid"), int) and entry["pid"] in holder_pids
         and spawner_is_dead(entry) is not False  # False = live Desktop supervisor owns it; keep refusing
     ]
 
@@ -377,27 +358,20 @@ def _serve_relaunch_commands(entries: list[dict]) -> list[list[str]]:
     (joined argv cannot round-trip Windows paths with spaces). Entries without a port are skipped.
     """
     from hermes_cli.update_cmd import _m
-    commands: list[list[str]] = []
     hermes = "hermes"
     with suppress(Exception):
         scripts_dir = _m()._venv_scripts_dir()
         if scripts_dir is not None:
-            for name in ("hermes.exe", "hermes"):
-                if (scripts_dir / name).is_file():
-                    hermes = str(scripts_dir / name)
-                    break
+            hermes = next((str(scripts_dir / n) for n in ("hermes.exe", "hermes") if (scripts_dir / n).is_file()), hermes)
+    commands: list[list[str]] = []
     for entry in entries:
         port = entry.get("port")
         if not isinstance(port, int) or port <= 0:
             continue
-        profile = str(entry.get("profile") or "")
-        host = str(entry.get("host") or "")
+        profile, host = str(entry.get("profile") or ""), str(entry.get("host") or "")
         commands.append(
-            [hermes]
-            + (["--profile", profile] if profile and profile != "default" else [])
-            + [str(entry.get("purpose"))]
-            + (["--host", host] if host else [])
-            + ["--port", str(port)]
+            [hermes] + (["--profile", profile] if profile and profile != "default" else [])
+            + [str(entry.get("purpose"))] + (["--host", host] if host else []) + ["--port", str(port)]
         )
     return commands
 
@@ -458,16 +432,15 @@ def _orphaned_desktop_backend_pids(matches: list[tuple[int, str, str]]) -> list[
     psutil = _psutil()
     if psutil is None:
         return None
-
     # Pass 1: find orphaned backend ROOTS among the holders.
     roots: list[tuple[int, int]] = []
-    remaining: list[tuple[int, str]] = []  # (pid, argv_low) still to justify
+    remaining: list[int] = []  # holders still to justify
     for pid, _name, cmdline in matches:
         low = _live_argv_low(psutil, pid, cmdline)
         if low is None:
             continue  # exited between scan and classification — nothing to reap
         if not _is_backend_argv(low):
-            remaining.append((int(pid), low))
+            remaining.append(int(pid))
             continue
         try:
             proc = psutil.Process(int(pid))
@@ -478,26 +451,24 @@ def _orphaned_desktop_backend_pids(matches: list[tuple[int, str, str]]) -> list[
             continue  # exited during classification — nothing to reap
         except Exception:
             return None
-
         try:
             ppid = proc.ppid()
             parent = psutil.Process(ppid) if ppid else None
-            # PID-reuse check: a "parent" created after its child is a recycled PID.
+            # PID-reuse check: a "parent" created after its child is a recycled PID. A live parent is
+            # not a root but may be an orphan root's descendant (the venv trampoline re-execs uv
+            # python with the SAME argv) — defer to pass 2.
             if parent is not None and parent.is_running() and parent.create_time() <= proc.create_time():
-                # Live parent: not a root, maybe an orphan root's descendant (the venv
-                # trampoline re-execs uv python with the SAME argv). Defer to pass 2.
-                remaining.append((int(pid), low))
+                remaining.append(int(pid))
                 continue
         except psutil.NoSuchProcess:
             pass  # parent gone → orphan
         except Exception:
             return None
         roots.append((int(pid), process_start_time))
-
     # Pass 2: every non-backend holder must descend from an accepted orphan root
     # (dies with the tree reap); anything else keeps the refusal.
     root_set = {pid for pid, _start_time in roots}
-    for pid, _low in remaining:
+    for pid in remaining:
         if not root_set:
             return None
         try:
@@ -519,17 +490,13 @@ def _ledger_reapable_backend_pids(matches: list[tuple[int, str, str]]) -> list[i
     """
     try:
         from hermes_cli.process_identity import REAPABLE_PURPOSES, ledger_entries, spawner_is_dead
-
         entries = ledger_entries()
     except Exception:
         return []
     by_pid = {e.get("pid"): e for e in entries if isinstance(e.get("pid"), int)}
     return [
-        int(pid)
-        for pid, _name, _cmdline in matches
-        if (entry := by_pid.get(int(pid)))
-        and entry.get("purpose") in REAPABLE_PURPOSES
-        and spawner_is_dead(entry) is True
+        int(pid) for pid, _name, _cmdline in matches
+        if (entry := by_pid.get(int(pid))) and entry.get("purpose") in REAPABLE_PURPOSES and spawner_is_dead(entry) is True
     ]
 
 
@@ -562,7 +529,6 @@ def _stop_process_trees(pids: list[int] | list[tuple[int, int]]) -> None:
     """
     from gateway.status import get_process_start_time
     from hermes_cli._subprocess_compat import pid_is_hermes, windows_hide_flags
-
     for entry in pids:
         pid, expected_start_time = entry if isinstance(entry, tuple) else (int(entry), get_process_start_time(int(entry)))
         try:
@@ -588,34 +554,32 @@ def _looks_like_desktop_control_plane(cmdline: str) -> bool:
     parser-derived classifier, never substring (``kanban --preserve-cache``, ``-m dashboard chat``).
     Undeterminable subcommand is NOT a control plane.
     """
-    return "hermes_cli.main" in (cmdline or "").lower() and _hermes_holder_subcommand(cmdline) in ("serve", "dashboard")
+    return "hermes_cli.main" in (cmdline or "").lower() and _hermes_holder_subcommand(cmdline) in _BACKEND_PURPOSES
 
 
 def _desktop_owns_gateway_lifecycle() -> bool:
     """True when Desktop currently supervises this install's control plane (updater must not steal gateway start).
 
     Not proof messaging is served: serve is the control plane, the gateway a detached sibling. Prefer the spawn
-    ledger; fall back to the venv-holder scan. An orphaned control plane (supervisor gone) does not count.
+    ledger; fall back to the venv-holder scan. An orphaned control plane (supervisor gone) does not count;
+    without psutil orphanhood is unprovable and a live control plane suffices.
     """
     from hermes_cli.update_cmd import _m
     with _best_effort('Desktop-lifecycle ledger probe failed: %s'):
         from hermes_cli.process_identity import ledger_entries, spawner_is_dead
-
-        if any(e.get("purpose") in ("serve", "dashboard") and spawner_is_dead(e) is False for e in ledger_entries()):
+        if any(e.get("purpose") in _BACKEND_PURPOSES and spawner_is_dead(e) is False for e in ledger_entries()):
             return True
-
     psutil = _psutil()
     try:
         holders = _m()._detect_venv_python_processes()
     except Exception as exc:
         logger.debug("Desktop-lifecycle holder scan failed: %s", exc)
         return False
-
     for pid, _name, cmdline in holders:
         if not _looks_like_desktop_control_plane(cmdline):
             continue
         if psutil is None:
-            return True  # cannot prove orphanhood; a live control plane suffices
+            return True
         with suppress(Exception):
             if _parent_is_live(psutil.Process(int(pid))):
                 return True
@@ -675,19 +639,14 @@ def _verify_service_identities(psutil, name: str, service, expected_service_iden
         if current_service_pid != int(expected_service_identity[0]):
             raise RuntimeError(f"Windows service {name} SCM process identity changed before stop")
     for label, identity in (("service", expected_service_identity), ("gateway", expected_gateway_identity)):
-        if identity is None:
-            continue
-        pid, create_time = identity
-        if abs(_process_create_time(psutil, pid, label, "before stop") - float(create_time)) > 0.001:
+        if identity is not None and abs(_process_create_time(psutil, identity[0], label, "before stop") - float(identity[1])) > 0.001:
             raise RuntimeError(f"Windows {label} process identity changed before stop")
     if expected_service_identity is not None and expected_gateway_identity is not None:
-        service_pid = int(expected_service_identity[0])
-        gateway_pid = int(expected_gateway_identity[0])
         try:
-            ancestor_pids = {int(parent.pid) for parent in psutil.Process(gateway_pid).parents()}
+            ancestor_pids = {int(parent.pid) for parent in psutil.Process(int(expected_gateway_identity[0])).parents()}
         except Exception as exc:
             raise RuntimeError("Windows gateway ancestry is unavailable before service stop") from exc
-        if service_pid not in ancestor_pids:
+        if int(expected_service_identity[0]) not in ancestor_pids:
             raise RuntimeError(f"Windows gateway is no longer owned by service {name}")
 
 
@@ -698,9 +657,11 @@ def _stop_windows_gateway_service(
     expected_gateway_identity: tuple[int, float] | None = None,
     timeout: float = 30.0,
 ) -> None:
-    """Stop one verified Windows service and wait until SCM reports it down."""
-    import psutil  # noqa: PLC0415
+    """Stop one verified Windows service and wait until SCM reports it down.
 
+    Lingering matching-identity processes after SCM says stopped make venv mutation unsafe — fail closed.
+    """
+    import psutil  # noqa: PLC0415
     service = psutil.win_service_get(name)
     _verify_service_identities(psutil, name, service, expected_service_identity, expected_gateway_identity)
     _sc_exe("stop", name, service, "stopped")
@@ -711,16 +672,13 @@ def _stop_windows_gateway_service(
     if not _poll_until(lambda: service.status() == "stopped" and not _alive(), timeout):
         if service.status() != "stopped":
             raise RuntimeError(f"Windows service {name} did not stop within {timeout:.0f}s; venv mutation unsafe.")
-        # Lingering matching-identity processes make venv mutation unsafe — fail closed.
-        alive_after_stop = _alive()
-        if alive_after_stop:
+        if alive_after_stop := _alive():
             raise RuntimeError(f"Windows service {name} stopped but its process tree is still alive: {alive_after_stop}")
 
 
 def _start_windows_gateway_service(name: str, *, timeout: float = 30.0) -> None:
     """Start one previously paused Windows service and verify it is running."""
     import psutil  # noqa: PLC0415
-
     service = psutil.win_service_get(name)
     _sc_exe("start", name, service, "running")
     if not _poll_until(lambda: service.status() == "running", timeout):
@@ -731,18 +689,16 @@ def _restore_windows_gateway_service(name: str, *, timeout: float = 60.0) -> Non
     """Restore a service after an uncertain stop, including STOP_PENDING."""
     from hermes_cli.update_cmd import _start_windows_gateway_service
     import psutil  # noqa: PLC0415
-
     service = psutil.win_service_get(name)
-    deadline = _time.monotonic() + timeout
-    while _time.monotonic() < deadline:
+
+    def _settled() -> bool:
         status = service.status()
-        if status == "running":
-            return
         if status == "stopped":
             _start_windows_gateway_service(name)
-            return
-        _time.sleep(0.2)
-    raise RuntimeError(f"Windows service {name} did not reach a restorable state within {timeout:.0f}s")
+        return status in ("running", "stopped")
+
+    if not _poll_until(_settled, timeout):
+        raise RuntimeError(f"Windows service {name} did not reach a restorable state within {timeout:.0f}s")
 
 
 def _windows_cold_start_plan() -> dict | None:
@@ -754,14 +710,12 @@ def _windows_cold_start_plan() -> dict | None:
     the skip is ownership, not liveness.
     """
     from hermes_cli.update_cmd import _desktop_owns_gateway_lifecycle
-
     with _best_effort('Could not check Desktop gateway-lifecycle ownership before update: %s'):
         if _desktop_owns_gateway_lifecycle():
             logger.debug("Skipping Windows gateway cold-start plan: Desktop owns gateway lifecycle")
             return None
     with _best_effort('Could not check Windows gateway autostart state before update: %s'):
         from hermes_cli import gateway_windows
-
         if gateway_windows.is_installed():
             return {"resume_needed": True, "profiles": {}, "unmapped_pids": [], "unmapped": [], "cold_start_if_installed": True}
     return None
@@ -774,7 +728,6 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic
     services AND the already-paused ordinary gateways before re-raising.
     """
     from hermes_cli.update_cmd import _restore_windows_gateway_service, _stop_windows_gateway_service
-
     paused_services = []
     current_service_name = None
     try:
@@ -790,9 +743,7 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic
             current_service_name = None
         if paused_services:
             token.update(services=paused_services, expected_services=list(paused_services), restarted_services=[])
-            token["service_profiles"] = {
-                str(s.name): str(s.profile) for s in service_gateways if str(s.name) in paused_services
-            }
+            token["service_profiles"] = {str(s.name): str(s.profile) for s in service_gateways if str(s.name) in paused_services}
             print("  ✓ Paused Windows gateway service(s): " + ", ".join(paused_services))
         return token
     except Exception as exc:
@@ -813,9 +764,8 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic
 
 
 def _discover_windows_gateways():
-    """``(profile_processes, service_gateways, running_pids)`` for the pause; any indeterminate probe aborts."""
+    """``(profile_processes, service_gateways, service_gateway_pids, running_pids)`` for the pause; any indeterminate probe aborts."""
     from hermes_cli.gateway import find_gateway_pids, find_profile_gateway_processes, find_windows_gateway_services
-
     with _abort_on_error("Could not map Windows gateway PIDs to profiles"):
         profile_process_list = find_profile_gateway_processes(strict=True)
         profile_processes = {proc.pid: proc for proc in profile_process_list}
@@ -847,7 +797,6 @@ def _request_socket_pauses(running_pids, profile_processes, service_gateway_pids
         _write_update_planned_stop_marker(Path(proc.path), int(pid))
         try:
             from gateway.control_socket import pause_gateway_for_update
-
             ack = pause_gateway_for_update(Path(proc.path))
             if ack and (ack.get("pausing") or ack.get("already_stopping")):
                 socket_acks.append(ack)
@@ -879,25 +828,18 @@ def _pause_windows_gateways_for_update() -> dict | None:
     locked during ``git``/``uv``. Stop only PIDs the gateway discovery code identifies.
     """
     from hermes_cli.update_cmd import _m
-
     if not _m()._is_windows():
         return None
-
     with _abort_on_error("Could not prepare Windows gateway pause for update"):
         from gateway.status import get_process_start_time, terminate_pid
         from hermes_cli.gateway import _capture_gateway_argv
-
     profile_processes, service_gateways, service_gateway_pids, running_pids = _discover_windows_gateways()
     if not running_pids:
         return _windows_cold_start_plan()
-
     profiles, mapped_pids, socket_acks = _request_socket_pauses(running_pids, profile_processes, service_gateway_pids)
-
-    # Resolve venv-side launchers BEFORE draining: a dead worker's parent cannot be
-    # recovered (NoSuchProcess). The launcher keeps ``.pyd`` mapped and would trip the
-    # venv-holder guard after the gateway stopped; it is killed with the survivors.
+    # Resolve venv-side launchers BEFORE draining: a dead worker's parent cannot be recovered (NoSuchProcess).
+    # The launcher keeps ``.pyd`` mapped and would trip the venv-holder guard; it is killed with the survivors.
     launcher_pids = _m()._venv_launcher_ancestors(mapped_pids)
-
     print("→ Stopping Windows gateway process(es) before updating Hermes...")
     drain_timeout = _gateway_drain_timeout(socket_acks)
     survivors = _m()._wait_for_windows_update_gateway_exit(mapped_pids, timeout=drain_timeout)
@@ -913,7 +855,6 @@ def _pause_windows_gateways_for_update() -> dict | None:
     # Snapshot unmapped gateways' argv *before* force-killing so resume can replay it.
     # Unmapped = no profile->PID-file mapping (e.g. Scheduled Task ``pythonw.exe -m ...``).
     unmapped = [{"pid": int(pid), "argv": _argv_or_none(int(pid))} for pid in unmapped_pids]
-
     # Tree-kill survivors, unmapped gateways, and pre-drain launchers; a launcher
     # already gone with its worker raises ProcessLookupError and is skipped.
     force_killed = []
@@ -921,17 +862,14 @@ def _pause_windows_gateways_for_update() -> dict | None:
         with suppress(ProcessLookupError, PermissionError, OSError):
             terminate_pid(int(pid), force=True, expected_start_time=get_process_start_time(int(pid)))
             force_killed.append(int(pid))
-
     if profiles:
         print(f"  ✓ Paused gateway profile(s): {', '.join(sorted(profiles))}")
     if force_killed:
         print(f"  → Force-stopped {len(force_killed)} gateway process(es)")
     if unmapped_pids:
         print(f"  → Stopped {len(unmapped_pids)} gateway process(es) without profile mapping")
-        if any(not u.get("argv") for u in unmapped):
-            # No recoverable cmdline (psutil missing, access denied, gone): manual restart.
+        if any(not u.get("argv") for u in unmapped):  # no recoverable cmdline (psutil missing, denied, gone)
             print("    Restart manually after update: hermes gateway run")
-
     token = {"resume_needed": True, "profiles": profiles, "unmapped_pids": unmapped_pids, "unmapped": unmapped}
     return _pause_windows_gateway_services(service_gateways, token, profiles, unmapped)
 
@@ -941,6 +879,7 @@ def _cold_start_windows_gateway_after_update() -> bool:
 
     Idempotent: re-checks nothing is running so a concurrent autostart can't duplicate. A successful Popen
     doesn't prove survival (a job object denying breakaway kills it), so success is gated on the liveness poll.
+    Vouched PIDs are attested so a death AFTER updater exit is reported by the next CLI invocation.
     """
     from hermes_cli.update_cmd import _desktop_owns_gateway_lifecycle, _m
     if not _m()._is_windows():
@@ -963,7 +902,6 @@ def _cold_start_windows_gateway_after_update() -> bool:
     if not ready_pids:
         raise RuntimeError(f"Windows gateway cold-start PID {pid} did not become ready")
     print(f"\n✓ Gateway started via cold-start after update (PID: {', '.join(map(str, ready_pids))})")
-    # Vouched PIDs: a death AFTER updater exit is reported by the next CLI invocation.
     with suppress(Exception):
         gateway_windows._write_start_attestation(ready_pids, "cold-start after update")
     return True
@@ -980,7 +918,6 @@ def _refresh_windows_gateway_launchers() -> None:
         return
     with _best_effort('Could not refresh Windows gateway launchers after update: %s'):
         from hermes_cli import gateway_windows
-
         if gateway_windows.is_installed():
             gateway_windows._write_task_script()
             print("  ✓ Refreshed Windows gateway launcher scripts")
@@ -998,10 +935,8 @@ def _refresh_bootstrap_cache_scripts(branch: str = "main") -> None:
         cache_dir = Path(_m().get_hermes_home()) / "bootstrap-cache"
         if not cache_dir.is_dir():
             return
-        # Mirror install_script.rs::sanitize_ref().
-        safe_ref = re.sub(r"[^A-Za-z0-9._-]", "_", str(branch or "main"))
-        # Mirror install_script.rs::is_valid_commit(): immutable commit pin, never rewrite.
-        if re.fullmatch(r"[0-9a-fA-F]{7,40}", safe_ref):
+        safe_ref = re.sub(r"[^A-Za-z0-9._-]", "_", str(branch or "main"))  # install_script.rs::sanitize_ref()
+        if re.fullmatch(r"[0-9a-fA-F]{7,40}", safe_ref):  # install_script.rs::is_valid_commit(): immutable pin
             return
         refreshed = []
         for kind, src_name in (("ps1", "install.ps1"), ("sh", "install.sh")):
@@ -1011,10 +946,9 @@ def _refresh_bootstrap_cache_scripts(branch: str = "main") -> None:
                 continue  # this ref was never bootstrap-cached — nothing to heal
             data = src.read_bytes()
             if kind == "ps1" and not data.startswith(b"\xef\xbb\xbf"):
-                # PowerShell needs the BOM or localized/em-dash text mis-decodes.
-                data = b"\xef\xbb\xbf" + data
+                data = b"\xef\xbb\xbf" + data  # PowerShell needs the BOM or localized/em-dash text mis-decodes.
             if cached.read_bytes() == data:
-                continue  # already current
+                continue
             tmp = cached.with_suffix(cached.suffix + ".tmp")
             tmp.write_bytes(data)
             os.replace(tmp, cached)
@@ -1031,22 +965,20 @@ def _resume_windows_services(token: dict) -> None:
     verified_restarts = list(token.get("restarted_services") or [])
     restarted_services = []
     failed_services = []
-    for service_name in services:
+    for service_name in map(str, services):
         try:
-            _start_windows_gateway_service(str(service_name))
-            restarted_services.append(str(service_name))
-            if str(service_name) not in verified_restarts:
-                verified_restarts.append(str(service_name))
+            _start_windows_gateway_service(service_name)
+            restarted_services.append(service_name)
+            if service_name not in verified_restarts:
+                verified_restarts.append(service_name)
         except Exception as exc:
             logger.warning("Could not restart Windows gateway service %s after update: %s", service_name, exc)
             print(f"  ⚠ Could not restart Windows gateway service: {service_name}")
-            failed_services.append(str(service_name))
-
+            failed_services.append(service_name)
     token["restarted_services"] = verified_restarts
+    token["services"] = failed_services
     if failed_services:
-        token["services"] = failed_services
         raise RuntimeError("Could not restart Windows gateway service(s): " + ", ".join(failed_services))
-    token["services"] = []
     if restarted_services:
         print("\n  ✓ Restarted Windows gateway service(s): " + ", ".join(restarted_services))
 
@@ -1071,27 +1003,21 @@ def _relaunch_paused_gateways(token: dict, profiles: dict, unmapped: list) -> tu
     relaunched = []
     failed_profiles = {}
     for profile, old_pid in sorted(profiles.items()):
-        if _launched(
-            lambda p=profile, o=old_pid: launch_detached_profile_gateway_restart(str(p), int(o)),
-            "Could not restart Windows gateway profile %s after update: %s", profile,
-        ):
+        if _launched(lambda p=profile, o=old_pid: launch_detached_profile_gateway_restart(str(p), int(o)),
+                     "Could not restart Windows gateway profile %s after update: %s", profile):
             relaunched.append(str(profile))
         else:
             failed_profiles[str(profile)] = int(old_pid)
     token["relaunched_profiles"] = relaunched
-
     unmapped_relaunched = 0
     failed_unmapped = []
     for entry in unmapped:
         argv, old_pid = entry.get("argv"), entry.get("pid")
-        if argv and old_pid and _launched(
-            lambda o=old_pid, a=argv: launch_detached_gateway_restart_by_cmdline(int(o), list(a)),
-            "Could not restart unmapped Windows gateway (pid %s) after update: %s", old_pid,
-        ):
+        if argv and old_pid and _launched(lambda o=old_pid, a=argv: launch_detached_gateway_restart_by_cmdline(int(o), list(a)),
+                                          "Could not restart unmapped Windows gateway (pid %s) after update: %s", old_pid):
             unmapped_relaunched += 1
         else:
             failed_unmapped.append(entry)
-
     token["profiles"] = failed_profiles
     token["unmapped"] = failed_unmapped
     if failed_profiles or failed_unmapped:
@@ -1132,12 +1058,10 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None:
     if not _m()._is_windows():
         token["resume_needed"] = False
         return
-
     # Regenerate launcher scripts before respawning so a legacy pythonw-era
     # autostart entry comes back on the current design at next login too.
     _m()._refresh_windows_gateway_launchers()
     _resume_windows_services(token)
-
     profiles = token.get("profiles") or {}
     unmapped = token.get("unmapped") or []
     if not profiles and not any(u.get("argv") for u in unmapped):
@@ -1147,12 +1071,10 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None:
             token["cold_start_if_installed"] = False
         token["resume_needed"] = False
         return
-
     relaunched, unmapped_relaunched = _relaunch_paused_gateways(token, profiles, unmapped)
     if relaunched or unmapped_relaunched:
         _verify_relaunched_gateways_alive(token, profiles, unmapped)
     token["resume_needed"] = False
-
     if relaunched:
         print(f"\n  ✓ Restarting Windows gateway profile(s): {', '.join(relaunched)}")
     if unmapped_relaunched:
@@ -1173,33 +1095,32 @@ def _resume_windows_gateways_and_merge_outcome(outcome, _windows_gateway_resume,
         print(f"  ⚠ Windows gateway service restart incomplete: {_windows_resume_exc}")
         if gateway_mode:
             _write_gateway_update_exit_code(False)
+    if not isinstance(_windows_gateway_resume, dict):
+        return
 
-    if isinstance(_windows_gateway_resume, dict):
-        def _extend_unique(target: list, items) -> None:
-            target.extend(item for item in dict.fromkeys(items) if item not in target)
+    def _extend_unique(target: list, items) -> None:
+        target.extend(item for item in dict.fromkeys(items) if item not in target)
 
-        token = _windows_gateway_resume
-        # Failed relaunches are absent from the token so they still surface. Best-effort.
-        with _best_effort('Could not merge Windows relaunch outcome into fleet reconciliation bookkeeping: %s'):
-            _extend_unique(outcome.relaunched_profiles, token.get("relaunched_profiles") or [])
-        windows_restarted = list(token.get("restarted_services") or [])
-        service_profiles = token.get("service_profiles") or {}
-        _extend_unique(outcome.restarted_services, windows_restarted)
-        _extend_unique(outcome.relaunched_profiles, (p for p in (service_profiles.get(n) for n in windows_restarted) if p))
-        _extend_unique(outcome.failed_or_stale_units, (str(service_profiles.get(n) or n) for n in (token.get("services") or [])))
-
-        with suppress(Exception):
-            from hermes_cli.update_receipt import record_gateway_restart
-
-            record_gateway_restart(
-                restarted_services=outcome.restarted_services,
-                relaunched_profiles=outcome.relaunched_profiles,
-                externally_supervised_profiles=outcome.externally_supervised_profiles,
-                killed_pids=sorted(outcome.killed_pids),
-                failed_units=outcome.failed_or_stale_units,
-                incomplete=outcome.incomplete or bool(outcome.failed_or_stale_units),
-                phase_error="; ".join(outcome.phase_errors) or None,
-            )
+    token = _windows_gateway_resume
+    # Failed relaunches are absent from the token so they still surface. Best-effort.
+    with _best_effort('Could not merge Windows relaunch outcome into fleet reconciliation bookkeeping: %s'):
+        _extend_unique(outcome.relaunched_profiles, token.get("relaunched_profiles") or [])
+    windows_restarted = list(token.get("restarted_services") or [])
+    service_profiles = token.get("service_profiles") or {}
+    _extend_unique(outcome.restarted_services, windows_restarted)
+    _extend_unique(outcome.relaunched_profiles, (p for p in (service_profiles.get(n) for n in windows_restarted) if p))
+    _extend_unique(outcome.failed_or_stale_units, (str(service_profiles.get(n) or n) for n in (token.get("services") or [])))
+    with suppress(Exception):
+        from hermes_cli.update_receipt import record_gateway_restart
+        record_gateway_restart(
+            restarted_services=outcome.restarted_services,
+            relaunched_profiles=outcome.relaunched_profiles,
+            externally_supervised_profiles=outcome.externally_supervised_profiles,
+            killed_pids=sorted(outcome.killed_pids),
+            failed_units=outcome.failed_or_stale_units,
+            incomplete=outcome.incomplete or bool(outcome.failed_or_stale_units),
+            phase_error="; ".join(outcome.phase_errors) or None,
+        )
 
 
 def _reap_and_rescan(message: str, pids, stop=None) -> list[tuple[int, str, str]]:
@@ -1214,7 +1135,6 @@ def _reap_and_rescan(message: str, pids, stop=None) -> list[tuple[int, str, str]
 def _terminate_leftover_gateways(pids) -> None:
     """Force-stop leftover gateways one by one; a failure is logged, never raised."""
     from gateway.status import get_process_start_time, terminate_pid
-
     for _pid in pids:
         try:
             terminate_pid(int(_pid), force=True, expected_start_time=get_process_start_time(int(_pid)))
@@ -1245,59 +1165,52 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew
     refused (the sync would corrupt against a locked .pyd).
     """
     from hermes_cli.update_cmd import _m, _record_update_step, _refuse_gateway_ancestor_tree_kill
-    _venv_holders = _m()._detect_venv_python_processes()
-    if _venv_holders:
-        # Gateways the pause machinery owns (respawned in the pause->guard window or
-        # unmapped spawn path): stop and re-check; post-update resume brings them back.
-        _gateway_holders = _m()._leftover_pausable_gateway_pids(_venv_holders)
-        if _gateway_holders is not None:
-            if _refuse_gateway_ancestor_tree_kill(_gateway_holders, gateway_mode=gateway_mode):
-                _m()._resume_windows_gateways_after_update(_windows_gateway_resume)
-                sys.exit(2)
-            _venv_holders = _reap_and_rescan(
-                f"  ⚠ {len(_gateway_holders)} gateway process(es) still hold the venv after the pause; stopping them",
-                _gateway_holders, stop=_terminate_leftover_gateways,
-            )
-    # Tree-reap rungs: (classifier, message). Ledger rung = positive identity in any context (self-registered
-    # backend, spawner provably dead; no PPID archaeology). Orphan rung = Desktop `serve` whose app is GONE
-    # (nothing respawns an orphan); live-Desktop backends return None and keep the refusal.
+
+    def _resume_and_exit():
+        _m()._resume_windows_gateways_after_update(_windows_gateway_resume)
+        sys.exit(2)
+
+    holders = _m()._detect_venv_python_processes()
+    # Gateways the pause machinery owns (respawned in the pause->guard window or unmapped
+    # spawn path): stop and re-check; post-update resume brings them back.
+    if holders and (gateway_holders := _m()._leftover_pausable_gateway_pids(holders)) is not None:
+        if _refuse_gateway_ancestor_tree_kill(gateway_holders, gateway_mode=gateway_mode):
+            _resume_and_exit()
+        holders = _reap_and_rescan(
+            f"  ⚠ {len(gateway_holders)} gateway process(es) still hold the venv after the pause; stopping them",
+            gateway_holders, stop=_terminate_leftover_gateways,
+        )
+    # Tree-reap rungs. Ledger rung = positive identity in any context (self-registered backend, spawner
+    # provably dead; no PPID archaeology). Orphan rung = Desktop `serve` whose app is GONE (nothing
+    # respawns an orphan); live-Desktop backends return None and keep the refusal.
     for classifier, message in (
         (_m()._ledger_reapable_backend_pids, "ledger-identified orphaned Hermes backend process(es) hold the venv"),
         (_m()._orphaned_desktop_backend_pids, "orphaned Desktop backend process(es) still hold the venv"),
     ):
-        if _venv_holders:
-            backends = classifier(_venv_holders)
-            if backends:
-                _venv_holders = _reap_and_rescan(f"  ⚠ {len(backends)} {message}; stopping their trees", backends)
-    if _venv_holders:
-        # Manual serve/dashboard rung (e.g. `hermes serve --host <ip>` for a REMOTE Desktop):
-        # ledger identity only (spawner dead; Desktop-owned keep the refusal). Stop and
-        # register an idempotent atexit relaunch on the SAME host/port/profile — success or failure.
-        _serve_entries = _m()._ledger_manual_serve_holders(_venv_holders)
-        if _serve_entries:
-            def _stop_and_park(pids):
-                _m()._stop_process_trees(pids)
-                _record_update_step("serve_pause", True, f"stopped={len(_serve_entries)}")
-                import atexit as _serve_atexit
+        if holders and (backends := classifier(holders)):
+            holders = _reap_and_rescan(f"  ⚠ {len(backends)} {message}; stopping their trees", backends)
+    # Manual serve/dashboard rung (e.g. `hermes serve --host <ip>` for a REMOTE Desktop): ledger identity
+    # only (spawner dead; Desktop-owned keep the refusal). Stop and register an idempotent atexit relaunch
+    # on the SAME host/port/profile — success or failure.
+    if holders and (serve_entries := _m()._ledger_manual_serve_holders(holders)):
+        def _stop_and_park(pids):
+            _m()._stop_process_trees(pids)
+            _record_update_step("serve_pause", True, f"stopped={len(serve_entries)}")
+            import atexit as _serve_atexit
+            _serve_atexit.register(_m()._relaunch_stopped_serves, {"pending": True, "entries": serve_entries})
 
-                _serve_atexit.register(_m()._relaunch_stopped_serves, {"pending": True, "entries": _serve_entries})
-
-            _venv_holders = _reap_and_rescan(
-                f"  ⚠ {len(_serve_entries)} manual serve/dashboard backend(s) hold the venv; stopping them for "
-                "the update (they will be relaunched on their recorded endpoints)",
-                [int(e["pid"]) for e in _serve_entries], stop=_stop_and_park,
-            )
-    if _venv_holders and _in_handoff_without_live_shim(args):
-        # Final rung: in a GUI hand-off the Desktop is contractually gone; surviving `serve`
-        # backends are leaks even with a live parent (which made the orphan-only rung bail
-        # and hang) — reap by cmdline.
-        _handoff_backends = _m()._handoff_reapable_backend_pids(_venv_holders)
-        if _handoff_backends:
-            _venv_holders = _reap_and_rescan(
-                f"  ⚠ {len(_handoff_backends)} Hermes backend process(es) "
-                "still hold the venv after the Desktop hand-off; stopping their trees", _handoff_backends,
-            )
-    if _venv_holders:
-        print(_format_venv_python_holders_message(_venv_holders))
-        _m()._resume_windows_gateways_after_update(_windows_gateway_resume)
-        sys.exit(2)
+        holders = _reap_and_rescan(
+            f"  ⚠ {len(serve_entries)} manual serve/dashboard backend(s) hold the venv; stopping them for "
+            "the update (they will be relaunched on their recorded endpoints)",
+            [int(e["pid"]) for e in serve_entries], stop=_stop_and_park,
+        )
+    # Final rung: in a GUI hand-off the Desktop is contractually gone; surviving `serve` backends are leaks
+    # even with a live parent (which made the orphan-only rung bail and hang) — reap by cmdline.
+    if holders and _in_handoff_without_live_shim(args) and (handoff_backends := _m()._handoff_reapable_backend_pids(holders)):
+        holders = _reap_and_rescan(
+            f"  ⚠ {len(handoff_backends)} Hermes backend process(es) "
+            "still hold the venv after the Desktop hand-off; stopping their trees", handoff_backends,
+        )
+    if holders:
+        print(_format_venv_python_holders_message(holders))
+        _resume_and_exit()

From 01d07b6b6cf510be9800f7da7dbe392516b93eeb Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:58:28 -0700
Subject: [PATCH 2/4] refactor(hermes_cli/update): unify try/log/default + SCM
 service handle in the Windows update module; compact docstrings

---
 hermes_cli/update_cmd_windows.py | 210 +++++++++++++------------------
 1 file changed, 85 insertions(+), 125 deletions(-)

diff --git a/hermes_cli/update_cmd_windows.py b/hermes_cli/update_cmd_windows.py
index 76def34f88..f69c13cd0e 100644
--- a/hermes_cli/update_cmd_windows.py
+++ b/hermes_cli/update_cmd_windows.py
@@ -22,6 +22,15 @@ logger = logging.getLogger("hermes_cli.update_cmd")  # log-record parity with th
 _BACKEND_PURPOSES = ("serve", "dashboard")
 
 
+def _try_call(fn, log_message: str, *log_args, default=None):
+    """``fn()``, or *default* after logging the exception at debug (``log_message`` gets ``*log_args, exc``)."""
+    try:
+        return fn()
+    except Exception as exc:
+        logger.debug(log_message, *log_args, exc)
+        return default
+
+
 @contextmanager
 def _abort_on_error(prefix: str):
     """Re-raise any failure of a mandatory step as ``RuntimeError(f"{prefix}: {exc}")`` (chained)."""
@@ -57,11 +66,13 @@ def _wait_for_windows_update_gateway_exit(pids: list[int], *, timeout: float) ->
             return False
 
     remaining = set(pids)
-    deadline = _time.monotonic() + max(timeout, 0.0)
-    while remaining and _time.monotonic() < deadline:
+
+    def _all_gone() -> bool:
+        nonlocal remaining
         remaining = {pid for pid in remaining if _alive(pid)}
-        if remaining:
-            _time.sleep(0.25)
+        return not remaining
+
+    _poll_until(_all_gone, max(timeout, 0.0), 0.25)
     return {pid for pid in remaining if _alive(pid)}
 
 
@@ -70,12 +81,10 @@ def _self_and_non_gateway_ancestor_pids(psutil) -> set[int]:
 
     Do NOT blanket-exclude ancestors: under ``/update`` the updater is a CHILD of the gateway, and hiding it
     dead-ends the update on ``venv-blocked``. Gateway ancestors stay visible (the pause path stops them
-    gracefully; a detached child survives on Windows); interactive ancestry is never a blocker.
-    """
-    try:
+    gracefully; a detached child survives on Windows); interactive ancestry is never a blocker."""
+    _is_gw = None
+    with suppress(Exception):
         from gateway.status import looks_like_gateway_command_line as _is_gw
-    except Exception:
-        _is_gw = None
     skip: set[int] = {os.getpid()}
     with suppress(Exception):
         for anc in psutil.Process().parents():
@@ -181,12 +190,11 @@ _holder_value_flags_cache: frozenset | None = None
 
 
 def _holder_value_flags() -> frozenset:
-    """Top-level CLI flags that consume a value, introspected from the REAL parser (nargs != 0); cached per process.
+    """Top-level CLI flags that consume a value, from the REAL parser (nargs != 0); cached per process.
 
     Derived so the holder classifier can't drift from argparse (a handwritten subset misparsed ``--reasoning high
     serve``). Pre-argparse profile selectors are added explicitly (stripped before argparse sees argv). Falls back
-    to a static snapshot when the parser can't import — the updater must classify holders even on a broken tree.
-    """
+    to a static snapshot when the parser can't import — the updater must classify holders even on a broken tree."""
     global _holder_value_flags_cache
     if _holder_value_flags_cache is not None:
         return _holder_value_flags_cache
@@ -206,8 +214,7 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None:
     """The actual Hermes SUBCOMMAND a venv-holder argv runs, or None (callers must NOT guess a label).
 
     Token-based, never substring (``kanban --preserve-cache`` contains "serve"): find the ``hermes_cli.main`` /
-    ``hermes(.exe)`` entry token, return the first following token that isn't a flag or a flag's value.
-    """
+    ``hermes(.exe)`` entry token, return the first following token that isn't a flag or a flag's value."""
     try:
         tokens = shlex.split(cmdline, posix=False)
     except Exception:
@@ -239,8 +246,7 @@ def _format_venv_python_holders_message(matches: list[tuple[int, str, str]]) ->
     """Explain which venv processes block the update and how to clear them.
 
     Labels come from the parsed SUBCOMMAND, never substring: a standalone ``hermes dashboard`` must not be
-    called the Desktop backend, ``--preserve-cache`` must not match "serve". Unknown argv gets no hint.
-    """
+    called the Desktop backend, ``--preserve-cache`` must not match "serve". Unknown argv gets no hint."""
     hint_by_subcommand = {
         "serve": "  ← Hermes backend (if the Desktop app is open, close it)",
         "dashboard": "  ← hermes dashboard (stop it: hermes dashboard stop, or close that terminal)",
@@ -252,14 +258,12 @@ def _format_venv_python_holders_message(matches: list[tuple[int, str, str]]) ->
         lines.append(f"  PID {pid}  {name}  {cmdline[:120]}{hint}")
     if len(matches) > 6:
         lines.append(f"  ... and {len(matches) - 6} more")
-    lines += [
-        "",
-        "  On Windows these keep native extension files (.pyd) locked, so the",
-        "  dependency update would fail partway and leave a broken install.",
-        "  Close the Hermes desktop app / other Hermes terminals, then re-run:",
-        "    hermes update",
-        "  (or use `hermes update --force-venv` to proceed anyway at your own risk)",
-    ]
+    lines.append(
+        "\n  On Windows these keep native extension files (.pyd) locked, so the\n"
+        "  dependency update would fail partway and leave a broken install.\n"
+        "  Close the Hermes desktop app / other Hermes terminals, then re-run:\n    hermes update\n"
+        "  (or use `hermes update --force-venv` to proceed anyway at your own risk)"
+    )
     return "\n".join(lines)
 
 
@@ -268,8 +272,7 @@ def _venv_launcher_ancestors(pids: list[int]) -> list[int]:
 
     A shim-started gateway is a chain: ``venv\\Scripts\\python.exe`` launcher (keeps ``.pyd`` mapped) -> uv
     CPython worker (writes the PID file). The pause set sees the worker, the venv scan sees the launcher, so a
-    paused gateway still tripped the guard. One hop up only, venv-prefixed only (bounds blast radius).
-    """
+    paused gateway still tripped the guard. One hop up only, venv-prefixed only (bounds blast radius)."""
     from hermes_cli.update_cmd import _m
     psutil = _psutil()
     if not _m()._is_windows() or not pids or psutil is None:
@@ -293,8 +296,7 @@ def _leftover_pausable_gateway_pids(matches: list[tuple[int, str, str]]) -> list
 
     A gateway respawned inside the pause->guard window (or via an unmapped spawn path) still holds ``.pyd`` files.
     Uses the Desktop preflight's ``_is_pausable_gateway`` so exemption and tolerance cannot drift; live argv is
-    re-read via psutil when possible since the scan may hold only a cmdline prefix.
-    """
+    re-read via psutil when possible since the scan may hold only a cmdline prefix."""
     from hermes_cli._scan_venv_blockers import _is_pausable_gateway
     psutil = _psutil()
     pids: list[int] = []
@@ -312,24 +314,21 @@ def _leftover_pausable_gateway_pids(matches: list[tuple[int, str, str]]) -> list
 def _refuse_gateway_ancestor_tree_kill(pids: list[int], *, gateway_mode: bool) -> bool:
     """Refuse a plain Windows update that would tree-kill its own ancestry (a chat agent's ``hermes update`` is
     a gateway child; ``taskkill /T /F`` kills the updater first). ``--gateway`` is exempt (detached delivery).
-    Refuse only when a nominated gateway is positively an ancestor; unknown ancestry keeps existing recovery.
-    """
+    Refuse only when a nominated gateway is positively an ancestor; unknown ancestry keeps existing recovery."""
     if gateway_mode or not pids:
         return False
-    try:
+    def _ancestors():
         from hermes_cli.gateway import _is_pid_ancestor_of_current_process
-        ancestors = [int(pid) for pid in pids if _is_pid_ancestor_of_current_process(int(pid))]
-    except Exception as exc:
-        logger.debug("Could not inspect gateway ancestry before tree-kill: %s", exc)
-        return False
+        return [int(pid) for pid in pids if _is_pid_ancestor_of_current_process(int(pid))]
+
+    ancestors = _try_call(_ancestors, "Could not inspect gateway ancestry before tree-kill: %s")
     if not ancestors:
         return False
     print(
         "✗ Refusing to stop the gateway process tree because this updater "
         f"is running inside it (gateway PID(s): {', '.join(str(pid) for pid in ancestors)}).\n"
         "  On Windows, taskkill /T would terminate the updater before the update can run.\n"
-        "  From a chat platform, use `/update` instead.\n"
-        "  Otherwise, run `hermes update` from a separate terminal."
+        "  From a chat platform, use `/update` instead.\n  Otherwise, run `hermes update` from a separate terminal."
     )
     return True
 
@@ -339,8 +338,7 @@ def _ledger_manual_serve_holders(matches: list[tuple[int, str, str]]) -> list[di
 
     Positive identity only: self-registered purpose serve/dashboard, live (pid, create_time), recorded spawner
     NOT alive (a Desktop-owned backend keeps its live Electron spawner and must keep the refusal — the app would
-    respawn what we kill). Full entries let the relauncher rebuild from host/port/profile, not argv.
-    """
+    respawn what we kill). Full entries let the relauncher rebuild from host/port/profile, not argv."""
     try:
         from hermes_cli.process_identity import ledger_entries, spawner_is_dead
     except Exception:
@@ -355,8 +353,7 @@ def _ledger_manual_serve_holders(matches: list[tuple[int, str, str]]) -> list[di
 
 def _serve_relaunch_commands(entries: list[dict]) -> list[list[str]]:
     """Rebuild launch commands for stopped serves from ledger host/port/profile — never argv parsing
-    (joined argv cannot round-trip Windows paths with spaces). Entries without a port are skipped.
-    """
+    (joined argv cannot round-trip Windows paths with spaces). Entries without a port are skipped."""
     from hermes_cli.update_cmd import _m
     hermes = "hermes"
     with suppress(Exception):
@@ -379,8 +376,7 @@ def _serve_relaunch_commands(entries: list[dict]) -> list[list[str]]:
 def _relaunch_stopped_serves(token: dict) -> None:
     """Idempotent atexit relaunch of manual serves stopped by the venv guard.
 
-    `pending` flips False on first invocation so explicit call + atexit registration cannot double-spawn.
-    """
+    `pending` flips False on first invocation so explicit call + atexit registration cannot double-spawn."""
     from hermes_cli.update_cmd import _m, _record_update_step
     if not token.get("pending"):
         return
@@ -395,8 +391,7 @@ def _relaunch_stopped_serves(token: dict) -> None:
         print("  ⟲ Relaunching stopped serve/dashboard backend(s)")
         failed = _m()._respawn_dashboard_processes(commands)
     if skipped or failed:
-        print("  ⚠ Some stopped backends could not be relaunched automatically; "
-              "restart them manually (hermes serve --host <ip> --port <port>).")
+        print("  ⚠ Some stopped backends could not be relaunched automatically; restart them manually (hermes serve --host <ip> --port <port>).")
     _record_update_step(
         "serve_relaunch", not failed and not skipped,
         f"relaunched={len(commands) - len(failed)} failed={len(failed)} skipped={skipped}",
@@ -427,8 +422,7 @@ def _orphaned_desktop_backend_pids(matches: list[tuple[int, str, str]]) -> list[
     would dead-end the update with "Hermes is still running" and zero open windows. Qualifies only if cmdline
     is a Hermes backend AND the parent is demonstrably gone (PID missing or reused). Tree-aware: holders inside
     an accepted root's tree fold into it; only roots are returned (``taskkill /T`` reaps descendants). Any
-    live-parent backend, unjustified non-backend, unprovable case, or no psutil -> ``None``. Never raises.
-    """
+    live-parent backend, unjustified non-backend, unprovable case, or no psutil -> ``None``. Never raises."""
     psutil = _psutil()
     if psutil is None:
         return None
@@ -486,8 +480,7 @@ def _ledger_reapable_backend_pids(matches: list[tuple[int, str, str]]) -> list[i
 
     Strongest rung (no PPID/cmdline inference): qualifies when ``(pid, create_time)`` matches a live ledger entry
     (PID reuse can't forge it), purpose is a REAPABLE kind (never interactive), and the recorded SPAWNER is
-    provably dead. Safe in ANY context. Unlisted holders fall to later rungs and never disqualify identified ones.
-    """
+    provably dead. Safe in ANY context. Unlisted holders fall to later rungs and never disqualify identified ones."""
     try:
         from hermes_cli.process_identity import REAPABLE_PURPOSES, ledger_entries, spawner_is_dead
         entries = ledger_entries()
@@ -506,8 +499,7 @@ def _handoff_reapable_backend_pids(matches: list[tuple[int, str, str]]) -> list[
     The orphan-only rung bails on ANY live parent (mid-teardown Electron, launcher->worker chain) and hung a
     hand-off. Inside the hand-off gate (marker + ``--gateway`` + no live ``hermes.exe`` shim) nothing legitimate
     supervises a ``serve`` from this venv, so survivors are leaks. Any non-backend holder or no psutil ->
-    ``None``. The CALLER must have confirmed the gate; outside it the stricter orphan-only path stands.
-    """
+    ``None``. The CALLER must have confirmed the gate; outside it the stricter orphan-only path stands."""
     psutil = _psutil()
     if psutil is None:
         return None
@@ -525,8 +517,7 @@ def _handoff_reapable_backend_pids(matches: list[tuple[int, str, str]]) -> list[
 def _stop_process_trees(pids: list[int] | list[tuple[int, int]]) -> None:
     """Force-stop each PID with its full child tree (Windows); best effort, never raises.
 
-    ``taskkill /T /F``: stopping only the parent can leave a ``.hermes-runtime`` child holding the install open.
-    """
+    ``taskkill /T /F``: stopping only the parent can leave a ``.hermes-runtime`` child holding the install open."""
     from gateway.status import get_process_start_time
     from hermes_cli._subprocess_compat import pid_is_hermes, windows_hide_flags
     for entry in pids:
@@ -552,8 +543,7 @@ def _looks_like_desktop_control_plane(cmdline: str) -> bool:
 
     Not the messaging gateway — don't feed into ``looks_like_gateway_command_line``. Token-based via the
     parser-derived classifier, never substring (``kanban --preserve-cache``, ``-m dashboard chat``).
-    Undeterminable subcommand is NOT a control plane.
-    """
+    Undeterminable subcommand is NOT a control plane."""
     return "hermes_cli.main" in (cmdline or "").lower() and _hermes_holder_subcommand(cmdline) in _BACKEND_PURPOSES
 
 
@@ -562,20 +552,14 @@ def _desktop_owns_gateway_lifecycle() -> bool:
 
     Not proof messaging is served: serve is the control plane, the gateway a detached sibling. Prefer the spawn
     ledger; fall back to the venv-holder scan. An orphaned control plane (supervisor gone) does not count;
-    without psutil orphanhood is unprovable and a live control plane suffices.
-    """
+    without psutil orphanhood is unprovable and a live control plane suffices."""
     from hermes_cli.update_cmd import _m
     with _best_effort('Desktop-lifecycle ledger probe failed: %s'):
         from hermes_cli.process_identity import ledger_entries, spawner_is_dead
         if any(e.get("purpose") in _BACKEND_PURPOSES and spawner_is_dead(e) is False for e in ledger_entries()):
             return True
     psutil = _psutil()
-    try:
-        holders = _m()._detect_venv_python_processes()
-    except Exception as exc:
-        logger.debug("Desktop-lifecycle holder scan failed: %s", exc)
-        return False
-    for pid, _name, cmdline in holders:
+    for pid, _name, cmdline in _try_call(_m()._detect_venv_python_processes, "Desktop-lifecycle holder scan failed: %s") or []:
         if not _looks_like_desktop_control_plane(cmdline):
             continue
         if psutil is None:
@@ -586,6 +570,12 @@ def _desktop_owns_gateway_lifecycle() -> bool:
     return False
 
 
+def _win_service(name: str):
+    """``(psutil, service)`` for the named SCM service (psutil imported here so tests can stub the module)."""
+    import psutil  # noqa: PLC0415
+    return psutil, psutil.win_service_get(name)
+
+
 def _sc_exe(verb: str, name: str, service, settled_status: str) -> None:
     """``sc.exe <verb> <name>``; a non-zero exit is only an error when SCM doesn't already report *settled_status*."""
     result = subprocess.run(
@@ -619,11 +609,12 @@ def _poll_until(condition, timeout: float, interval: float = 0.2) -> bool:
     return False
 
 
-def _process_create_time(psutil, pid: int, label: str, when: str) -> float:
+def _process_create_time(psutil, pid: int, label: str) -> float:
+    """``create_time`` of *pid*; unreadable identity aborts the stop (RuntimeError)."""
     try:
         return float(psutil.Process(int(pid)).create_time())
     except Exception as exc:
-        raise RuntimeError(f"Windows {label} process identity is unavailable {when}") from exc
+        raise RuntimeError(f"Windows {label} process identity is unavailable before stop") from exc
 
 
 def _verify_service_identities(psutil, name: str, service, expected_service_identity, expected_gateway_identity) -> None:
@@ -639,7 +630,7 @@ def _verify_service_identities(psutil, name: str, service, expected_service_iden
         if current_service_pid != int(expected_service_identity[0]):
             raise RuntimeError(f"Windows service {name} SCM process identity changed before stop")
     for label, identity in (("service", expected_service_identity), ("gateway", expected_gateway_identity)):
-        if identity is not None and abs(_process_create_time(psutil, identity[0], label, "before stop") - float(identity[1])) > 0.001:
+        if identity is not None and abs(_process_create_time(psutil, identity[0], label) - float(identity[1])) > 0.001:
             raise RuntimeError(f"Windows {label} process identity changed before stop")
     if expected_service_identity is not None and expected_gateway_identity is not None:
         try:
@@ -659,10 +650,8 @@ def _stop_windows_gateway_service(
 ) -> None:
     """Stop one verified Windows service and wait until SCM reports it down.
 
-    Lingering matching-identity processes after SCM says stopped make venv mutation unsafe — fail closed.
-    """
-    import psutil  # noqa: PLC0415
-    service = psutil.win_service_get(name)
+    Lingering matching-identity processes after SCM says stopped make venv mutation unsafe — fail closed."""
+    psutil, service = _win_service(name)
     _verify_service_identities(psutil, name, service, expected_service_identity, expected_gateway_identity)
     _sc_exe("stop", name, service, "stopped")
 
@@ -678,8 +667,7 @@ def _stop_windows_gateway_service(
 
 def _start_windows_gateway_service(name: str, *, timeout: float = 30.0) -> None:
     """Start one previously paused Windows service and verify it is running."""
-    import psutil  # noqa: PLC0415
-    service = psutil.win_service_get(name)
+    service = _win_service(name)[1]
     _sc_exe("start", name, service, "running")
     if not _poll_until(lambda: service.status() == "running", timeout):
         raise RuntimeError(f"Windows service {name} did not start within {timeout:.0f}s")
@@ -688,8 +676,7 @@ def _start_windows_gateway_service(name: str, *, timeout: float = 30.0) -> None:
 def _restore_windows_gateway_service(name: str, *, timeout: float = 60.0) -> None:
     """Restore a service after an uncertain stop, including STOP_PENDING."""
     from hermes_cli.update_cmd import _start_windows_gateway_service
-    import psutil  # noqa: PLC0415
-    service = psutil.win_service_get(name)
+    service = _win_service(name)[1]
 
     def _settled() -> bool:
         status = service.status()
@@ -707,8 +694,7 @@ def _windows_cold_start_plan() -> dict | None:
     An installed autostart entry is an explicit "I want a gateway" signal; a gateway that died between
     updates would otherwise stay down until next login (resume only relaunches what was running).
     Desktop-owned lifecycle -> ``None`` (spawning ``gateway run`` beside Desktop races ports/state);
-    the skip is ownership, not liveness.
-    """
+    the skip is ownership, not liveness."""
     from hermes_cli.update_cmd import _desktop_owns_gateway_lifecycle
     with _best_effort('Could not check Desktop gateway-lifecycle ownership before update: %s'):
         if _desktop_owns_gateway_lifecycle():
@@ -725,8 +711,7 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic
     """Stop each SCM gateway service, recording them on *token*; roll everything back on failure.
 
     Runs after every fallible ordinary-gateway step so a failure here restores the attempted
-    services AND the already-paused ordinary gateways before re-raising.
-    """
+    services AND the already-paused ordinary gateways before re-raising."""
     from hermes_cli.update_cmd import _restore_windows_gateway_service, _stop_windows_gateway_service
     paused_services = []
     current_service_name = None
@@ -783,8 +768,7 @@ def _request_socket_pauses(running_pids, profile_processes, service_gateway_pids
     """Marker + socket-first pause for every profile-mapped gateway; ``(profiles, mapped_pids, socket_acks)``.
 
     Socket ACK = the gateway drains and exits by its own graceful path. No answer (older
-    gateway) -> the marker poll / force-kill ladder in the caller.
-    """
+    gateway) -> the marker poll / force-kill ladder in the caller."""
     profiles: dict[str, int] = {}
     mapped_pids = []
     socket_acks: list[dict] = []
@@ -825,8 +809,7 @@ def _pause_windows_gateways_for_update() -> dict | None:
     """Stop running Windows gateways before mutating the checkout or venv.
 
     Scheduled/startup gateways run via pythonw.exe, invisible to the hermes.exe instance guard, yet keep files
-    locked during ``git``/``uv``. Stop only PIDs the gateway discovery code identifies.
-    """
+    locked during ``git``/``uv``. Stop only PIDs the gateway discovery code identifies."""
     from hermes_cli.update_cmd import _m
     if not _m()._is_windows():
         return None
@@ -844,17 +827,13 @@ def _pause_windows_gateways_for_update() -> dict | None:
     drain_timeout = _gateway_drain_timeout(socket_acks)
     survivors = _m()._wait_for_windows_update_gateway_exit(mapped_pids, timeout=drain_timeout)
     unmapped_pids = [pid for pid in running_pids if pid not in profile_processes and pid not in service_gateway_pids]
-
-    def _argv_or_none(pid: int):
-        try:
-            return _capture_gateway_argv(pid)
-        except Exception as exc:
-            logger.debug("Could not capture argv for unmapped gateway %s: %s", pid, exc)
-            return None
-
     # Snapshot unmapped gateways' argv *before* force-killing so resume can replay it.
     # Unmapped = no profile->PID-file mapping (e.g. Scheduled Task ``pythonw.exe -m ...``).
-    unmapped = [{"pid": int(pid), "argv": _argv_or_none(int(pid))} for pid in unmapped_pids]
+    unmapped = [
+        {"pid": int(pid), "argv": _try_call(lambda p=int(pid): _capture_gateway_argv(p),
+                                            "Could not capture argv for unmapped gateway %s: %s", int(pid))}
+        for pid in unmapped_pids
+    ]
     # Tree-kill survivors, unmapped gateways, and pre-drain launchers; a launcher
     # already gone with its worker raises ProcessLookupError and is skipped.
     force_killed = []
@@ -879,8 +858,7 @@ def _cold_start_windows_gateway_after_update() -> bool:
 
     Idempotent: re-checks nothing is running so a concurrent autostart can't duplicate. A successful Popen
     doesn't prove survival (a job object denying breakaway kills it), so success is gated on the liveness poll.
-    Vouched PIDs are attested so a death AFTER updater exit is reported by the next CLI invocation.
-    """
+    Vouched PIDs are attested so a death AFTER updater exit is reported by the next CLI invocation."""
     from hermes_cli.update_cmd import _desktop_owns_gateway_lifecycle, _m
     if not _m()._is_windows():
         return True
@@ -911,8 +889,7 @@ def _refresh_windows_gateway_launchers() -> None:
     """Regenerate installed Windows gateway launcher scripts after update; best-effort, never fails the update.
 
     Launchers are written once at install, so old installs kept launching via ``pythonw.exe`` (``sys.stderr is
-    None`` death). The task's /TR points at a stable path, so rewriting in place retargets it without UAC.
-    """
+    None`` death). The task's /TR points at a stable path, so rewriting in place retargets it without UAC."""
     from hermes_cli.update_cmd import _m
     if not _m()._is_windows():
         return
@@ -928,8 +905,7 @@ def _refresh_bootstrap_cache_scripts(branch: str = "main") -> None:
 
     Old ``hermes-setup.exe`` builds NEVER re-download a cached branch-ref script, so a stale one runs
     months-old code forever. Guards mirror ``install_script.rs``: only the sanitized *branch* key is rewritten;
-    commit-SHA pins (7-40 hex) are immutable and skipped. Best-effort: never fails the update.
-    """
+    commit-SHA pins (7-40 hex) are immutable and skipped. Best-effort: never fails the update."""
     from hermes_cli.update_cmd import _m
     with _best_effort('Could not refresh bootstrap-cache scripts after update: %s'):
         cache_dir = Path(_m().get_hermes_home()) / "bootstrap-cache"
@@ -987,23 +963,15 @@ def _relaunch_paused_gateways(token: dict, profiles: dict, unmapped: list) -> tu
     """Relaunch profile gateways and replay unmapped argv; ``(relaunched_profiles, unmapped_count)``.
 
     Failed relaunches stay on the token (and off ``relaunched_profiles``) so plan-vs-execution
-    reconciliation still surfaces them — Windows has no watcher to recover them.
-    """
+    reconciliation still surfaces them — Windows has no watcher to recover them."""
     with _abort_on_error("Could not load Windows gateway restart helper"):
         from hermes_cli.gateway import launch_detached_gateway_restart_by_cmdline, launch_detached_profile_gateway_restart
 
-    def _launched(launch, log_message: str, log_arg) -> bool:
-        """``launch()`` truthiness; an exception (incl. bad pid/argv coercion) logs at debug and reads False."""
-        try:
-            return bool(launch())
-        except Exception as exc:
-            logger.debug(log_message, log_arg, exc)
-            return False
-
+    # An exception from a launch (incl. bad pid/argv coercion) logs at debug and reads as a failed relaunch.
     relaunched = []
     failed_profiles = {}
     for profile, old_pid in sorted(profiles.items()):
-        if _launched(lambda p=profile, o=old_pid: launch_detached_profile_gateway_restart(str(p), int(o)),
+        if _try_call(lambda p=profile, o=old_pid: launch_detached_profile_gateway_restart(str(p), int(o)),
                      "Could not restart Windows gateway profile %s after update: %s", profile):
             relaunched.append(str(profile))
         else:
@@ -1013,7 +981,7 @@ def _relaunch_paused_gateways(token: dict, profiles: dict, unmapped: list) -> tu
     failed_unmapped = []
     for entry in unmapped:
         argv, old_pid = entry.get("argv"), entry.get("pid")
-        if argv and old_pid and _launched(lambda o=old_pid, a=argv: launch_detached_gateway_restart_by_cmdline(int(o), list(a)),
+        if argv and old_pid and _try_call(lambda o=old_pid, a=argv: launch_detached_gateway_restart_by_cmdline(int(o), list(a)),
                                           "Could not restart unmapped Windows gateway (pid %s) after update: %s", old_pid):
             unmapped_relaunched += 1
         else:
@@ -1030,8 +998,7 @@ def _verify_relaunched_gateways_alive(token: dict, profiles: dict, unmapped: lis
 
     A parent Job Object denying CREATE_BREAKAWAY_FROM_JOB can kill the gateway on updater teardown;
     ``all_profiles=True`` covers the fleet. Vouched PIDs are persisted so a death AFTER updater exit
-    is reported by the next CLI invocation (best-effort).
-    """
+    is reported by the next CLI invocation (best-effort)."""
     with _abort_on_error("Could not load Windows gateway liveness helpers"):
         from hermes_cli import gateway_windows
     ready_pids = gateway_windows._wait_for_gateway_ready(timeout_s=30.0, all_profiles=True)
@@ -1039,10 +1006,8 @@ def _verify_relaunched_gateways_alive(token: dict, profiles: dict, unmapped: lis
         token["profiles"] = dict(profiles)
         token["unmapped"] = list(unmapped)
         print(
-            "\n  ⚠ Windows gateway restart could not be verified — no stable "
-            "gateway process appeared after relaunch.\n"
-            "    (The respawned gateway may have been killed by a parent "
-            "Job Object during updater teardown, #48820.)\n"
+            "\n  ⚠ Windows gateway restart could not be verified — no stable gateway process appeared after relaunch.\n"
+            "    (The respawned gateway may have been killed by a parent Job Object during updater teardown, #48820.)\n"
             "    Recover with: hermes gateway restart"
         )
         raise RuntimeError("Windows gateway relaunch after update was not verified alive")
@@ -1084,8 +1049,7 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None:
 
 def _resume_windows_gateways_and_merge_outcome(outcome, _windows_gateway_resume, gateway_mode: bool):
     """Resume gateways paused for a Windows update and fold the token into ``outcome``'s systemd/launchd-style
-    bookkeeping so reconciliation never reports a healthy gateway as unaccounted. Must never abort the update.
-    """
+    bookkeeping so reconciliation never reports a healthy gateway as unaccounted. Must never abort the update."""
     from hermes_cli.update_cmd import _m, _write_gateway_update_exit_code
     try:
         _m()._resume_windows_gateways_after_update(_windows_gateway_resume)
@@ -1136,17 +1100,14 @@ def _terminate_leftover_gateways(pids) -> None:
     """Force-stop leftover gateways one by one; a failure is logged, never raised."""
     from gateway.status import get_process_start_time, terminate_pid
     for _pid in pids:
-        try:
-            terminate_pid(int(_pid), force=True, expected_start_time=get_process_start_time(int(_pid)))
-        except Exception as exc:
-            logger.debug("Could not stop leftover gateway %s: %s", _pid, exc)
+        _try_call(lambda p=int(_pid): terminate_pid(p, force=True, expected_start_time=get_process_start_time(p)),
+                  "Could not stop leftover gateway %s: %s", _pid)
 
 
 def _in_handoff_without_live_shim(args) -> bool:
     """GUI hand-off gate: ``--gateway`` + update-incomplete marker AND no live ``hermes.exe`` shim.
 
-    Fail closed: unverifiable marker or shim state reads as "not a hand-off" / "live shim".
-    """
+    Fail closed: unverifiable marker or shim state reads as "not a hand-off" / "live shim"."""
     from hermes_cli.update_cmd import _m
     try:
         if not (bool(getattr(args, "gateway", False)) and _m()._update_marker_path().exists()):
@@ -1162,8 +1123,7 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew
 
     Rungs in order: leftover pausable gateways -> ledger orphaned backends -> orphaned Desktop backends ->
     ledger manual serve (relaunched at exit on the same bind) -> GUI hand-off leaks. Remaining holders are
-    refused (the sync would corrupt against a locked .pyd).
-    """
+    refused (the sync would corrupt against a locked .pyd)."""
     from hermes_cli.update_cmd import _m, _record_update_step, _refuse_gateway_ancestor_tree_kill
 
     def _resume_and_exit():

From d1bb7fb9cf0802a120da6a2cb2fa0b6c0e677600 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Thu, 3 Sep 2026 00:02:04 -0700
Subject: [PATCH 3/4] refactor(hermes_cli/update): fold venv-holder
 classification and service-stop guards in the Windows update module

---
 hermes_cli/update_cmd_windows.py | 75 +++++++++++++++-----------------
 1 file changed, 35 insertions(+), 40 deletions(-)

diff --git a/hermes_cli/update_cmd_windows.py b/hermes_cli/update_cmd_windows.py
index f69c13cd0e..acefb6a5d2 100644
--- a/hermes_cli/update_cmd_windows.py
+++ b/hermes_cli/update_cmd_windows.py
@@ -45,9 +45,12 @@ def _write_update_planned_stop_marker(profile_path: Path, pid: int) -> bool:
     try:
         from gateway.status import _get_process_start_time
         from utils import atomic_json_write
-        record = {"target_pid": pid, "target_start_time": _get_process_start_time(pid),
-                  "stopper_pid": os.getpid(), "written_at": datetime.now(timezone.utc).isoformat()}
-        atomic_json_write(Path(profile_path) / ".gateway-planned-stop.json", record, indent=None, separators=(",", ":"))
+        atomic_json_write(
+            Path(profile_path) / ".gateway-planned-stop.json",
+            {"target_pid": pid, "target_start_time": _get_process_start_time(pid), "stopper_pid": os.getpid(),
+             "written_at": datetime.now(timezone.utc).isoformat()},
+            indent=None, separators=(",", ":"),
+        )
         return True
     except (OSError, PermissionError):
         return False
@@ -160,28 +163,30 @@ def _detect_venv_python_processes(*, exclude_pids: set[int] | None = None) -> li
         except (OSError, ValueError):
             exe_norm = str(exe).lower()
         # Primary match: exe lives under this venv (desktop backend / gateway case).
-        is_holder = exe_norm.startswith(venv_prefix)
+        in_venv = exe_norm.startswith(venv_prefix)
         name = str(info.get("name") or Path(exe).name)
         name_low = name.lower()
-        if not is_holder and not (name_low.startswith(("python", "pypy")) or name_low in {"uv.exe", "uvx.exe", "hermes.exe"}):
+        if not (in_venv or name_low.startswith(("python", "pypy")) or name_low in {"uv.exe", "uvx.exe", "hermes.exe"}):
             continue
         cmdline_raw = _cmdline_or_empty(proc)
         cmdline_low = cmdline_raw.lower()
         # Fallback: uv/base-interpreter trampolines have an exe OUTSIDE the venv yet hold
         # its .pyd files — match cmdline (venv path, or `-m hermes_cli.main` + root/cwd).
-        if not is_holder:
-            is_holder = venv_prefix in cmdline_low
-        if not is_holder and "hermes_cli.main" in cmdline_low:
-            try:
-                cwd_low = str(proc.cwd() or "").lower().rstrip(os.sep) + os.sep
-            except Exception:
-                cwd_low = os.sep
-            is_holder = root_prefix in cmdline_low or cwd_low.startswith(root_prefix)
-        if is_holder:
+        if in_venv or venv_prefix in cmdline_low or (
+            "hermes_cli.main" in cmdline_low and (root_prefix in cmdline_low or _cwd_prefix(proc).startswith(root_prefix))
+        ):
             matches.append((int(pid), name, cmdline_raw))
     return matches
 
 
+def _cwd_prefix(proc) -> str:
+    """Lower-cased cwd of *proc* with one trailing separator; bare ``os.sep`` when unreadable."""
+    try:
+        return str(proc.cwd() or "").lower().rstrip(os.sep) + os.sep
+    except Exception:
+        return os.sep
+
+
 _HOLDER_VALUE_FLAGS_FALLBACK = frozenset({
     "--profile", "-p", "--config", "--model", "-m", "--provider", "--reasoning", "--toolsets", "-t",
     "--skills", "-s", "--continue", "-c", "--resume", "-r", "--oneshot", "-z", "--in", "--usage-file",
@@ -222,9 +227,8 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None:
 
     def _is_entry(i: int, token: str) -> bool:
         low = token.lower().strip('"')
-        if low.endswith("hermes_cli.main") and i > 0 and tokens[i - 1] == "-m":
-            return True
-        return low.rsplit("\\", 1)[-1].rsplit("/", 1)[-1] in ("hermes", "hermes.exe")
+        return (low.endswith("hermes_cli.main") and i > 0 and tokens[i - 1] == "-m") or (
+            low.rsplit("\\", 1)[-1].rsplit("/", 1)[-1] in ("hermes", "hermes.exe"))
 
     entry_idx = next((i for i, token in enumerate(tokens) if _is_entry(i, token)), None)
     if entry_idx is None:
@@ -463,10 +467,8 @@ def _orphaned_desktop_backend_pids(matches: list[tuple[int, str, str]]) -> list[
     # (dies with the tree reap); anything else keeps the refusal.
     root_set = {pid for pid, _start_time in roots}
     for pid in remaining:
-        if not root_set:
-            return None
         try:
-            if not root_set & {int(a.pid) for a in psutil.Process(pid).parents()}:
+            if not root_set or not root_set & {int(a.pid) for a in psutil.Process(pid).parents()}:
                 return None
         except psutil.NoSuchProcess:
             continue  # exited already
@@ -621,8 +623,7 @@ def _verify_service_identities(psutil, name: str, service, expected_service_iden
     """Refuse to stop *name* unless SCM state, service/gateway process identities and ancestry all still match."""
     if expected_service_identity is not None:
         try:
-            current_status = str(service.status())
-            current_service_pid = int(service.pid() or 0)
+            current_status, current_service_pid = str(service.status()), int(service.pid() or 0)
         except Exception as exc:
             raise RuntimeError(f"Windows service {name} SCM identity is unavailable before stop") from exc
         if current_status != "running":
@@ -642,11 +643,8 @@ def _verify_service_identities(psutil, name: str, service, expected_service_iden
 
 
 def _stop_windows_gateway_service(
-    name: str, *,
-    expected_processes: tuple[tuple[int, float], ...] = (),
-    expected_service_identity: tuple[int, float] | None = None,
-    expected_gateway_identity: tuple[int, float] | None = None,
-    timeout: float = 30.0,
+    name: str, *, expected_processes: tuple[tuple[int, float], ...] = (), expected_service_identity: tuple[int, float] | None = None,
+    expected_gateway_identity: tuple[int, float] | None = None, timeout: float = 30.0,
 ) -> None:
     """Stop one verified Windows service and wait until SCM reports it down.
 
@@ -658,11 +656,12 @@ def _stop_windows_gateway_service(
     def _alive() -> list[int]:
         return [pid for pid, create_time in expected_processes if _original_process_is_alive(psutil, pid, create_time)]
 
-    if not _poll_until(lambda: service.status() == "stopped" and not _alive(), timeout):
-        if service.status() != "stopped":
-            raise RuntimeError(f"Windows service {name} did not stop within {timeout:.0f}s; venv mutation unsafe.")
-        if alive_after_stop := _alive():
-            raise RuntimeError(f"Windows service {name} stopped but its process tree is still alive: {alive_after_stop}")
+    if _poll_until(lambda: service.status() == "stopped" and not _alive(), timeout):
+        return
+    if service.status() != "stopped":
+        raise RuntimeError(f"Windows service {name} did not stop within {timeout:.0f}s; venv mutation unsafe.")
+    if alive_after_stop := _alive():
+        raise RuntimeError(f"Windows service {name} stopped but its process tree is still alive: {alive_after_stop}")
 
 
 def _start_windows_gateway_service(name: str, *, timeout: float = 30.0) -> None:
@@ -719,8 +718,7 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic
         for service in service_gateways:
             current_service_name = str(service.name)
             _stop_windows_gateway_service(
-                current_service_name,
-                expected_processes=tuple(getattr(service, "descendant_identities", ())),
+                current_service_name, expected_processes=tuple(getattr(service, "descendant_identities", ())),
                 expected_service_identity=(int(service.service_pid), float(service.service_create_time)),
                 expected_gateway_identity=(int(service.gateway_pid), float(service.gateway_create_time)),
             )
@@ -1077,12 +1075,9 @@ def _resume_windows_gateways_and_merge_outcome(outcome, _windows_gateway_resume,
     with suppress(Exception):
         from hermes_cli.update_receipt import record_gateway_restart
         record_gateway_restart(
-            restarted_services=outcome.restarted_services,
-            relaunched_profiles=outcome.relaunched_profiles,
-            externally_supervised_profiles=outcome.externally_supervised_profiles,
-            killed_pids=sorted(outcome.killed_pids),
-            failed_units=outcome.failed_or_stale_units,
-            incomplete=outcome.incomplete or bool(outcome.failed_or_stale_units),
+            restarted_services=outcome.restarted_services, relaunched_profiles=outcome.relaunched_profiles,
+            externally_supervised_profiles=outcome.externally_supervised_profiles, killed_pids=sorted(outcome.killed_pids),
+            failed_units=outcome.failed_or_stale_units, incomplete=outcome.incomplete or bool(outcome.failed_or_stale_units),
             phase_error="; ".join(outcome.phase_errors) or None,
         )
 

From b93fe9cfaa083dbb6d96773170a603944c51cc8e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Thu, 3 Sep 2026 00:07:26 -0700
Subject: [PATCH 4/4] =?UTF-8?q?refactor(hermes=5Fcli/update):=20git=20modu?=
 =?UTF-8?q?le=20=E2=80=94=20=5Fgit=5Fstdout=20helper,=20fetch-failure=20ru?=
 =?UTF-8?q?le=20table,=20upstream-offer=20phase=20helper,=20single-pass=20?=
 =?UTF-8?q?EOL=20probe?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 hermes_cli/update_cmd_git.py | 348 ++++++++++++++---------------------
 1 file changed, 142 insertions(+), 206 deletions(-)

diff --git a/hermes_cli/update_cmd_git.py b/hermes_cli/update_cmd_git.py
index a5b0c4b640..fcd47cd989 100644
--- a/hermes_cli/update_cmd_git.py
+++ b/hermes_cli/update_cmd_git.py
@@ -13,23 +13,29 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Optional
 
-# Log-record parity with the origin module.
-logger = logging.getLogger("hermes_cli.update_cmd")
+logger = logging.getLogger("hermes_cli.update_cmd")  # log-record parity with the origin module
 
 _ORPHAN_RESCUE_REFS_TO_KEEP = 10
 _ORPHAN_RESCUE_REF_MAX_AGE_DAYS = 30
 
 _GIT_TEXT_KW = dict(capture_output=True, text=True, encoding="utf-8", errors="replace")
 _BAR = "=" * 68
+_UPSTREAM_ADD_CMD = "git remote add upstream https://github.com/NousResearch/hermes-agent.git"
 
 
 def _git_ok(git_cmd, args, cwd, **kw) -> bool:
     """True when ``_git_run`` exits 0; any exception counts as failure."""
+    return _git_stdout(git_cmd, args, cwd, **kw) is not None
+
+
+def _git_stdout(git_cmd, args, cwd, **kw) -> Optional[str]:
+    """Stripped stdout of a successful ``_git_run``; ``None`` on non-zero exit or any exception."""
     from hermes_cli.update_cmd import _git_run
-    try:
-        return _git_run(git_cmd, args, cwd, **kw).returncode == 0
-    except Exception:
-        return False
+    with suppress(Exception):
+        result = _git_run(git_cmd, args, cwd, **kw)
+        if result.returncode == 0:
+            return result.stdout.strip()
+    return None
 
 
 def _prune_orphan_rescue_refs(
@@ -37,11 +43,10 @@ def _prune_orphan_rescue_refs(
 ) -> None:
     """Expire old orphan rescue refs (``refs/hermes-update-backups/orphan-<branch>-<ts>-<sha>``).
 
-    Each ref pins a possibly multi-GB snapshot against ``git gc``, so a repeatedly corrupted
-    install would grow ``.git`` unbounded. Keep the ``keep`` newest AND drop any older than
-    ``max_age_days`` by the ``YYYYMMDD-HHMMSS`` stamp (unparseable names left alone); names
-    sort chronologically so ``for-each-ref`` order is creation order. Best-effort, never blocks.
-    """
+    Each ref pins a possibly multi-GB snapshot against ``git gc``, so a repeatedly corrupted install would
+    grow ``.git`` unbounded. Keep the ``keep`` newest AND drop any older than ``max_age_days`` by the
+    ``YYYYMMDD-HHMMSS`` stamp (unparseable names left alone); names sort chronologically so
+    ``for-each-ref`` order is creation order. Best-effort, never blocks."""
     from hermes_cli.update_cmd import _git_run
     with suppress(OSError):
         prefix = f"refs/hermes-update-backups/orphan-{branch}-"
@@ -53,22 +58,17 @@ def _prune_orphan_rescue_refs(
         if max_age_days > 0:
             cutoff = datetime.now(timezone.utc) - timedelta(days=max_age_days)
             for ref in refs:
-                stamp = ref[len(prefix):][:15]  # "YYYYMMDD-HHMMSS"
-                try:
-                    ref_time = datetime.strptime(stamp, "%Y%m%d-%H%M%S").replace(tzinfo=timezone.utc)
-                except ValueError:
-                    continue
-                if ref_time < cutoff:
-                    stale.add(ref)
+                with suppress(ValueError):
+                    if datetime.strptime(ref[len(prefix):][:15], "%Y%m%d-%H%M%S").replace(tzinfo=timezone.utc) < cutoff:
+                        stale.add(ref)
         for ref in sorted(stale):
             _git_run(git_cmd, ["update-ref", "-d", ref], cwd)
 
 
 def _branch_head_label(git_cmd=None, cwd=None) -> str | None:
-    """``"<branch> @ <short-sha>"`` for the checkout, or None when unknown. Never raises.
+    """``"<branch> @ <short-sha>"`` for the checkout (``detached`` when not on a branch), or None. Never raises.
 
-    Appended to summary lines so a checkout parked on a stale branch is visible.
-    """
+    Appended to summary lines so a checkout parked on a stale branch is visible."""
     from hermes_cli.update_cmd import _m
     try:
         cmd = list(git_cmd) if git_cmd else ["git"]
@@ -81,8 +81,7 @@ def _branch_head_label(git_cmd=None, cwd=None) -> str | None:
         branch_name, sha_text = branch.stdout.strip(), sha.stdout.strip()
         if branch.returncode != 0 or sha.returncode != 0 or not sha_text or not branch_name:
             return None
-        label = "detached" if branch_name == "HEAD" else branch_name
-        return f"{label} @ {sha_text}"
+        return f"{'detached' if branch_name == 'HEAD' else branch_name} @ {sha_text}"
     except Exception:
         return None
 
@@ -93,18 +92,16 @@ def _branch_head_suffix(git_cmd=None, cwd=None) -> str:
     return f" [{label}]" if label else ""
 
 
-def _assess_parked_branch_switch(
-    git_cmd: list[str], cwd: Path, current_branch: str, target_branch: str
-) -> tuple[bool, str]:
+def _assess_parked_branch_switch(git_cmd: list[str], cwd: Path, current_branch: str, target_branch: str) -> tuple[bool, str]:
     """Decide whether a parked feature branch may be auto-switched back to the update target.
 
     - (True, "") — tree clean and every parked commit is in ``origin/<target>`` (no ``git cherry +``).
-    - (True, "unmerged:<n>") — tree clean but commits not in target; switching is safe (checkout
-      keeps committed work) but caller must print a LOUD notice. Non-interactive callers
-      (desktop, gateway /update, cron) can't resolve a skip, so a clean checkout must reach target.
-    - (False, "disabled"|"dirty"|"unverifiable") — caller must NOT touch the branch. Dirty is
-      the genuinely unsafe case: uncommitted work riding an autostash across branches.
-    """
+    - (True, "unmerged:<n>") — tree clean but commits not in target; switching is safe (checkout keeps
+      committed work) but caller must print a LOUD notice. Non-interactive callers (desktop, gateway
+      /update, cron) can't resolve a skip, so a clean checkout must reach target.
+    - (False, "disabled"|"dirty"|"unverifiable") — caller must NOT touch the branch. Dirty is the
+      genuinely unsafe case: uncommitted work riding an autostash across branches.
+    A config read failure must not disable the safety checks: fall through with the default."""
     from hermes_cli.update_cmd import _git_run
     try:
         from hermes_cli.config import load_config
@@ -112,23 +109,17 @@ def _assess_parked_branch_switch(
         if isinstance(_update_cfg, dict) and not bool(_update_cfg.get("auto_switch_parked_branch", True)):
             return False, "disabled"
     except Exception as exc:
-        # Config read failure must not disable the safety checks; fall through with default.
         logger.debug("Could not read updates.auto_switch_parked_branch: %s", exc)
-
     status = _git_run(git_cmd, ["status", "--porcelain"], cwd)
     if status.returncode != 0:
         return False, "unverifiable"
     if status.stdout.strip():
         return False, "dirty"
-
     cherry = _git_run(git_cmd, ["cherry", f"origin/{target_branch}"], cwd)
     if cherry.returncode != 0:
         return False, "unverifiable"
     unmerged = [line for line in cherry.stdout.splitlines() if line.startswith("+")]
-    if unmerged:
-        # Safe (checkout keeps commits); reason tells caller to print the loud notice.
-        return True, f"unmerged:{len(unmerged)}"
-    return True, ""
+    return True, f"unmerged:{len(unmerged)}" if unmerged else ""
 
 
 _PARKED_SKIP_WHY = {
@@ -137,24 +128,14 @@ _PARKED_SKIP_WHY = {
 }
 
 
-def _print_parked_branch_skip_warning(
-    git_cmd: list[str],
-    cwd: Path,
-    current_branch: str,
-    target_branch: str,
-    reason: str,
-) -> None:
+def _print_parked_branch_skip_warning(git_cmd: list[str], cwd: Path, current_branch: str, target_branch: str, reason: str) -> None:
     """LOUD block: why the update was skipped on a parked branch, behind-count, fix commands."""
-    from hermes_cli.update_cmd import _git_run
     behind = None
     with suppress(Exception):
-        behind_result = _git_run(git_cmd, ["rev-list", f"HEAD..origin/{target_branch}", "--count"], cwd)
-        if behind_result.returncode == 0 and behind_result.stdout.strip():
-            behind = int(behind_result.stdout.strip())
-
-    why = _PARKED_SKIP_WHY.get(
-        reason, f"the branch state could not be verified against origin/{target_branch}"
-    )
+        behind_text = _git_stdout(git_cmd, ["rev-list", f"HEAD..origin/{target_branch}", "--count"], cwd)
+        if behind_text:
+            behind = int(behind_text)
+    why = _PARKED_SKIP_WHY.get(reason, f"the branch state could not be verified against origin/{target_branch}")
     print(f"\n{_BAR}\n⚠ CODE UPDATE SKIPPED — checkout is parked on '{current_branch}'")
     print(f"  Not auto-switching to {target_branch}: {why}.")
     if behind is not None and behind > 0:
@@ -167,14 +148,11 @@ def _print_parked_branch_skip_warning(
     )
 
 
-def _print_parked_branch_kept_notice(
-    current_branch: str, target_branch: str, unmerged_count: str
-) -> None:
+def _print_parked_branch_kept_notice(current_branch: str, target_branch: str, unmerged_count: str) -> None:
     """LOUD notice when a clean parked branch with unmerged commits is auto-switched.
 
     Non-interactive callers can't resolve a skip, so we proceed — but the unmerged work
-    (still safe on its branch) must be impossible to miss.
-    """
+    (still safe on its branch) must be impossible to miss."""
     print(
         f"\n{_BAR}\n"
         f"⚠ Checkout was parked on '{current_branch}' with "
@@ -197,12 +175,7 @@ SKIP_UPSTREAM_PROMPT_FILE = ".skip_upstream_prompt"
 
 def _get_origin_url(git_cmd: list[str], cwd: Path) -> Optional[str]:
     """Get the URL of the origin remote, or None if not set."""
-    from hermes_cli.update_cmd import _git_run
-    with suppress(Exception):
-        result = _git_run(git_cmd, ["remote", "get-url", "origin"], cwd)
-        if result.returncode == 0:
-            return result.stdout.strip()
-    return None
+    return _git_stdout(git_cmd, ["remote", "get-url", "origin"], cwd)
 
 
 def _is_fork(origin_url: Optional[str]) -> bool:
@@ -229,18 +202,16 @@ def _add_upstream_remote(git_cmd: list[str], cwd: Path) -> bool:
 
 def _count_commits_between(git_cmd: list[str], cwd: Path, base: str, head: str) -> int:
     """Count commits on `head` that are not on `base`. Returns -1 on error."""
-    from hermes_cli.update_cmd import _git_run
     with suppress(Exception):
-        result = _git_run(git_cmd, ["rev-list", "--count", f"{base}..{head}"], cwd)
-        if result.returncode == 0:
-            return int(result.stdout.strip())
+        count = _git_stdout(git_cmd, ["rev-list", "--count", f"{base}..{head}"], cwd)
+        if count is not None:
+            return int(count)
     return -1
 
 
 def _should_skip_upstream_prompt() -> bool:
     """Check if user previously declined to add upstream."""
     from hermes_constants import get_hermes_home
-
     return (get_hermes_home() / SKIP_UPSTREAM_PROMPT_FILE).exists()
 
 
@@ -248,7 +219,6 @@ def _mark_skip_upstream_prompt():
     """Create marker file to skip future upstream prompts."""
     with suppress(Exception):
         from hermes_constants import get_hermes_home
-
         (get_hermes_home() / SKIP_UPSTREAM_PROMPT_FILE).touch()
 
 
@@ -257,61 +227,55 @@ def _sync_fork_with_upstream(git_cmd: list[str], cwd: Path) -> bool:
     return _git_ok(git_cmd, ["push", "origin", "main", "--force-with-lease"], cwd, network=True)
 
 
-def _sync_with_upstream_if_needed(
-    git_cmd: list[str],
-    cwd: Path,
-    *,
-    assume_yes: bool = False,
-    input_fn=None,
-) -> bool:
-    """Offer to add ``upstream``, compare origin/main vs upstream/main, ff-pull when strictly
-    behind, then push origin.
+def _offer_upstream_remote(git_cmd: list[str], cwd: Path, *, assume_yes: bool, input_fn) -> bool:
+    """Prompt to add ``upstream`` and add it; False when the user declined, the run is non-interactive, or add failed.
 
-    Returns True only when origin/main was actually verified against upstream/main; False when
-    the check never happened, so the caller never reports "up to date" on an origin-only compare.
-    """
-    from hermes_cli.update_cmd import (
-        _add_upstream_remote, _count_commits_between, _has_upstream_remote,
-        _mark_skip_upstream_prompt, _no_prompt_git_kwargs, _should_skip_upstream_prompt,
+    ``--yes`` means "don't block", not "mutate my remotes", so a non-interactive skip is NOT persisted."""
+    from hermes_cli.update_cmd import _add_upstream_remote, _mark_skip_upstream_prompt
+    print(
+        "\nℹ Your fork is not tracking the official Hermes repository.\n"
+        "  This means you may miss updates from NousResearch/hermes-agent.\n"
     )
-    if not _has_upstream_remote(git_cmd, cwd):
-        if _should_skip_upstream_prompt():
-            return False
-        print(
-            "\nℹ Your fork is not tracking the official Hermes repository.\n"
-            "  This means you may miss updates from NousResearch/hermes-agent.\n"
-        )
-        if assume_yes or (input_fn is None and not (sys.stdin.isatty() and sys.stdout.isatty())):
-            # --yes means "don't block", not "mutate my remotes"; don't persist the decline.
-            print("  Skipping upstream setup (non-interactive run).\n"
-                  "  Add it later with: git remote add upstream https://github.com/NousResearch/hermes-agent.git")
-            return False
-        if input_fn is not None:
-            response = input_fn("Add official repo as 'upstream' remote? [y/N]", "n").strip().lower()
-        else:
-            try:
-                response = input("Add official repo as 'upstream' remote? [Y/n]: ").strip().lower()
-            except (EOFError, KeyboardInterrupt, UnicodeDecodeError):
-                print()
-                response = "n"
-        if response not in {"", "y", "yes"}:
-            print("  Skipped. Run 'git remote add upstream https://github.com/NousResearch/hermes-agent.git' to add later.")
-            _mark_skip_upstream_prompt()
-            return False
-        print("→ Adding upstream remote...")
-        if not _add_upstream_remote(git_cmd, cwd):
-            print("  ✗ Failed to add upstream remote. Skipping upstream sync.")
-            return False
-        print("  ✓ Added upstream: https://github.com/NousResearch/hermes-agent.git")
+    if assume_yes or (input_fn is None and not (sys.stdin.isatty() and sys.stdout.isatty())):
+        print(f"  Skipping upstream setup (non-interactive run).\n  Add it later with: {_UPSTREAM_ADD_CMD}")
+        return False
+    if input_fn is not None:
+        response = input_fn("Add official repo as 'upstream' remote? [y/N]", "n").strip().lower()
+    else:
+        try:
+            response = input("Add official repo as 'upstream' remote? [Y/n]: ").strip().lower()
+        except (EOFError, KeyboardInterrupt, UnicodeDecodeError):
+            print()
+            response = "n"
+    if response not in {"", "y", "yes"}:
+        print(f"  Skipped. Run '{_UPSTREAM_ADD_CMD}' to add later.")
+        _mark_skip_upstream_prompt()
+        return False
+    print("→ Adding upstream remote...")
+    if not _add_upstream_remote(git_cmd, cwd):
+        print("  ✗ Failed to add upstream remote. Skipping upstream sync.")
+        return False
+    print("  ✓ Added upstream: https://github.com/NousResearch/hermes-agent.git")
+    return True
 
-    # Only upstream/main: a bare fetch drags in thousands of auto-generated branches.
+
+def _sync_with_upstream_if_needed(git_cmd: list[str], cwd: Path, *, assume_yes: bool = False, input_fn=None) -> bool:
+    """Offer to add ``upstream``, compare origin/main vs upstream/main, ff-pull when strictly behind, then push origin.
+
+    Returns True only when origin/main was actually verified against upstream/main; False when the check never
+    happened, so the caller never reports "up to date" on an origin-only compare. Fetches only upstream/main:
+    a bare fetch drags in thousands of auto-generated branches."""
+    from hermes_cli.update_cmd import _count_commits_between, _has_upstream_remote, _no_prompt_git_kwargs, _should_skip_upstream_prompt
+    if not _has_upstream_remote(git_cmd, cwd) and (
+        _should_skip_upstream_prompt() or not _offer_upstream_remote(git_cmd, cwd, assume_yes=assume_yes, input_fn=input_fn)
+    ):
+        return False
     print("\n→ Fetching upstream...")
     try:
         subprocess.run(git_cmd + ["fetch", "upstream", "main", "--quiet"], cwd=cwd, capture_output=True, check=True, **_no_prompt_git_kwargs())
     except subprocess.CalledProcessError:
         print("  ✗ Failed to fetch upstream. Skipping upstream sync.")
         return False
-
     origin_ahead = _count_commits_between(git_cmd, cwd, "upstream/main", "origin/main")
     upstream_ahead = _count_commits_between(git_cmd, cwd, "origin/main", "upstream/main")
     if origin_ahead < 0 or upstream_ahead < 0:
@@ -327,14 +291,12 @@ def _sync_with_upstream_if_needed(
     if upstream_ahead == 0:
         print("  ✓ Fork is up to date with upstream")
         return True
-
     print(f"\n→ Fork is {upstream_ahead} commit(s) behind upstream\n→ Pulling from upstream...")
     try:
         subprocess.run(git_cmd + ["pull", "--ff-only", "upstream", "main"], cwd=cwd, check=True, **_no_prompt_git_kwargs())
     except subprocess.CalledProcessError:
         print("  ✗ Failed to pull from upstream. You may need to resolve conflicts manually.")
         return False
-
     print("  ✓ Updated from upstream\n→ Syncing fork...")
     if _sync_fork_with_upstream(git_cmd, cwd):
         print("  ✓ Fork synced with upstream")
@@ -346,33 +308,33 @@ def _sync_with_upstream_if_needed(
     return True
 
 
+def _has_http_code(stderr: str, *codes: str) -> bool:
+    return any(f"HTTP {code}" in stderr or f"returned error: {code}" in stderr for code in codes)
+
+
+# Ordered (predicate, diagnosis): curl reports HTTP errors as ``unable to access '<url>': ... error: 429``,
+# so rate-limit/outage checks must run BEFORE the generic "unable to access" check. An anonymous fetch
+# answered with HTTP 401 ("could not read Username") is GitHub during an outage (or a renamed/private
+# repo), not a user credentials problem.
+_FETCH_FAILURE_RULES = (
+    (lambda s: _has_http_code(s, "429") or "rate limit" in s.lower(),
+     "✗ GitHub is rate limiting requests or having an outage (HTTP 429) — try again in 5 minutes."),
+    (lambda s: _has_http_code(s, "500", "502", "503", "504"),
+     "✗ GitHub appears to be having an outage — try again in a few minutes (https://www.githubstatus.com)."),
+    (lambda s: "Could not resolve host" in s or "unable to access" in s,
+     "✗ Network error — cannot reach the remote repository."),
+    (lambda s: "could not read Username" in s or "terminal prompts disabled" in s,
+     "✗ GitHub rejected the anonymous fetch (asked for a login) — this usually means a GitHub outage;"
+     " try again in a few minutes (https://www.githubstatus.com). If it persists, check"
+     " `git remote -v` points at a public repo."),
+    (lambda s: "Authentication failed" in s,
+     "✗ Authentication failed — check your git credentials or SSH key."),
+)
+
+
 def _classify_fetch_failure(stderr: str) -> str:
-    """Map git-fetch stderr to a one-line diagnosis (caller also prints the raw first line).
-
-    Order matters: curl reports HTTP errors as ``unable to access '<url>': ... error: 429``,
-    so rate-limit/outage checks must run BEFORE the generic "unable to access" check.
-    """
-
-    def _has_http_code(*codes: str) -> bool:
-        return any(f"HTTP {code}" in stderr or f"returned error: {code}" in stderr for code in codes)
-
-    if _has_http_code("429") or "rate limit" in stderr.lower():
-        return "✗ GitHub is rate limiting requests or having an outage (HTTP 429) — try again in 5 minutes."
-    if _has_http_code("500", "502", "503", "504"):
-        return "✗ GitHub appears to be having an outage — try again in a few minutes (https://www.githubstatus.com)."
-    if "Could not resolve host" in stderr or "unable to access" in stderr:
-        return "✗ Network error — cannot reach the remote repository."
-    if "could not read Username" in stderr or "terminal prompts disabled" in stderr:
-        # Anonymous fetch got HTTP 401: GitHub does this during outages (and for
-        # renamed/private repos) — not a user credentials problem.
-        return (
-            "✗ GitHub rejected the anonymous fetch (asked for a login) — this usually means a GitHub outage;"
-            " try again in a few minutes (https://www.githubstatus.com). If it persists, check"
-            " `git remote -v` points at a public repo."
-        )
-    if "Authentication failed" in stderr:
-        return "✗ Authentication failed — check your git credentials or SSH key."
-    return "✗ Failed to fetch updates from origin."
+    """Map git-fetch stderr to a one-line diagnosis (caller also prints the raw first line)."""
+    return next((message for matches, message in _FETCH_FAILURE_RULES if matches(stderr)), "✗ Failed to fetch updates from origin.")
 
 
 def _print_fetch_failure(stderr: str) -> None:
@@ -395,16 +357,15 @@ def _probe_fork_bomb(argv: list) -> Optional[bool]:
 def _git_is_trampoline(git_cmd: list) -> bool:
     """Whether *git_cmd* is a broken Git-for-Windows trampoline shim.
 
-    The ~46KB ``bin\\git.exe``/``cmd\\git.exe`` shims re-exec git-core; when they can't find
-    it every call dies with the launcher's guard message (a PATH problem, not network).
-    Never raises; unknown states report False so a probe failure can't block an update.
-    """
+    The ~46KB ``bin\\git.exe``/``cmd\\git.exe`` shims re-exec git-core; when they can't find it every call
+    dies with the launcher's guard message (a PATH problem, not network). Never raises; unknown states
+    report False so a probe failure can't block an update."""
     return _probe_fork_bomb(git_cmd) is True
 
 
 def _portable_git_candidates() -> list:
-    """PortableGit candidates: shared root first (where the managed tree actually lives,
-    not the profile-scoped HERMES_HOME), then profile home as a fallback for custom layouts."""
+    """PortableGit candidates: shared root first (where the managed tree actually lives, not the
+    profile-scoped HERMES_HOME), then profile home as a fallback for custom layouts."""
     from hermes_cli.update_cmd import get_default_hermes_root, get_hermes_home
     candidates = []
     with suppress(Exception):
@@ -413,24 +374,20 @@ def _portable_git_candidates() -> list:
 
 
 def _locate_real_git() -> Optional[Path]:
-    """Find a real Git-for-Windows ``git-core/git.exe`` (standard locations + managed PortableGit)
-    that runs without the trampoline guard. None when nothing suits — callers keep the broken
-    command and let the fetch-failure ZIP fallback handle it."""
+    """Find a real Git-for-Windows ``git-core/git.exe`` (standard locations + managed PortableGit) that runs
+    without the trampoline guard. None when nothing suits — callers keep the broken command and let the
+    fetch-failure ZIP fallback handle it. A failed probe (None) disqualifies a candidate like a guard hit."""
     candidates = [
         Path(r"C:\Program Files\Git\mingw64\libexec\git-core\git.exe"),
         Path(r"C:\Program Files (x86)\Git\mingw64\libexec\git-core\git.exe"),
     ] + _portable_git_candidates()
-    for candidate in candidates:
-        # A failed probe (None) disqualifies the candidate just like a guard hit.
-        if candidate.exists() and _probe_fork_bomb([str(candidate)]) is False:
-            return candidate
-    return None
+    return next((c for c in candidates if c.exists() and _probe_fork_bomb([str(c)]) is False), None)
 
 
 def _ensure_non_trampoline_git(git_cmd: list) -> list:
-    """Swap a broken Git-for-Windows trampoline for a real git binary so fetch/pull/checkout
-    keep working; if none is found leave the command untouched (fetch-failure handler falls
-    back to ZIP). No-op off Windows and when git is healthy."""
+    """Swap a broken Git-for-Windows trampoline for a real git binary so fetch/pull/checkout keep working;
+    if none is found leave the command untouched (fetch-failure handler falls back to ZIP). No-op off
+    Windows and when git is healthy."""
     from hermes_cli.update_cmd import _locate_real_git
     if sys.platform != "win32" or not _git_is_trampoline(git_cmd):
         return git_cmd
@@ -443,9 +400,8 @@ def _ensure_non_trampoline_git(git_cmd: list) -> list:
 
 
 def _discard_lockfile_churn(git_cmd, repo_root):
-    """Restore ``package-lock.json`` files npm rewrote non-deterministically, so the update
-    sees a clean tree instead of autostashing every run. Only touches lockfiles whose
-    package.json is NOT also dirty. Best-effort."""
+    """Restore ``package-lock.json`` files npm rewrote non-deterministically, so the update sees a clean tree
+    instead of autostashing every run. Only touches lockfiles whose package.json is NOT also dirty. Best-effort."""
     from hermes_cli.update_cmd import _git_run
     with suppress(Exception):
         diff = _git_run(git_cmd, ["diff", "--name-only"], repo_root)
@@ -463,13 +419,12 @@ def _discard_lockfile_churn(git_cmd, repo_root):
 def _normalize_managed_eol(git_cmd, repo_root):
     """Take a managed checkout off ``core.autocrlf=true`` without leaving it dirty.
 
-    Git for Windows sets ``autocrlf=true`` system-wide, turning LF files CRLF and breaking
-    ``git checkout`` on update; install.ps1 pins ``false`` but older checkouts never got it
-    and only ``hermes update`` can fix them. Pin and cleanup are one operation: under
-    ``autocrlf=true`` a CRLF tree reads clean, so pinning alone would expose every file as
-    modified (whole-tree autostash). Pin only after the tree verifies clean under it; a
-    checkout we can't fully normalize is left as-is. Best-effort.
-    """
+    Git for Windows sets ``autocrlf=true`` system-wide, turning LF files CRLF and breaking ``git checkout``
+    on update; install.ps1 pins ``false`` but older checkouts never got it and only ``hermes update`` can
+    fix them. Pin and cleanup are one operation: under ``autocrlf=true`` a CRLF tree reads clean, so pinning
+    alone would expose every file as modified (whole-tree autostash). Pin only after the tree verifies clean
+    under it; a checkout we can't fully normalize is left as-is. Only ``true`` rewrites LF->CRLF
+    (unset/false/input leave the tree alone). Best-effort."""
     from hermes_cli.update_cmd import _git_run
     # -c, not config: evaluate the tree as it WOULD look pinned, persisting nothing.
     probe = git_cmd + ["-c", "core.autocrlf=false"]
@@ -477,51 +432,32 @@ def _normalize_managed_eol(git_cmd, repo_root):
     def _probe_run(*args, **kw):
         return subprocess.run(probe + list(args), cwd=repo_root, **_GIT_TEXT_KW, **kw)
 
-    def _dirty():
-        out = _probe_run("diff", "-z", "--name-only")
-        if out.returncode != 0:
+    def _eol_only():
+        """Dirty paths whose ONLY change is CRLF; None when either probe fails."""
+        all_dirty = _probe_run("diff", "-z", "--name-only")
+        # Files with a *content* change ignoring CRLF. ``--name-only --ignore-cr-at-eol`` still LISTS
+        # CR-only files; ``--numstat`` honors the filter (no record for them). Records are
+        # "<added>\\t<deleted>\\t<path>"; rename detection is off, so exactly one path.
+        real_dirty = _probe_run("-c", "core.quotepath=false", "diff", "--numstat", "--ignore-cr-at-eol")
+        if all_dirty.returncode != 0 or real_dirty.returncode != 0:
             return None
-        return {p for p in out.stdout.split("\0") if p}
-
-    def _real_dirty():
-        # Files with a *content* change ignoring CRLF. ``--name-only --ignore-cr-at-eol``
-        # still LISTS CR-only files; ``--numstat`` honors the filter (no record for them).
-        out = _probe_run("-c", "core.quotepath=false", "diff", "--numstat", "--ignore-cr-at-eol")
-        if out.returncode != 0:
-            return None
-        # "<added>\t<deleted>\t<path>"; rename detection off, so exactly one path.
-        return {
+        return {p for p in all_dirty.stdout.split("\0") if p} - {
             parts[2]
-            for parts in (line.split("\t", 2) for line in out.stdout.splitlines() if line.strip())
+            for parts in (line.split("\t", 2) for line in real_dirty.stdout.splitlines() if line.strip())
             if len(parts) == 3 and parts[2]
         }
 
-    def _eol_only():
-        all_dirty, real_dirty = _dirty(), _real_dirty()
-        if all_dirty is None or real_dirty is None:
-            return None
-        return all_dirty - real_dirty
-
     with suppress(Exception):
-        effective = _git_run(git_cmd, ["config", "--get", "core.autocrlf"], repo_root)
-        # Only "true" rewrites LF->CRLF; unset/false/input leave the tree alone.
-        if effective.stdout.strip().lower() != "true":
+        if _git_run(git_cmd, ["config", "--get", "core.autocrlf"], repo_root).stdout.strip().lower() != "true":
             return
-
         eol_only = _eol_only()
         if eol_only is None:
             return
         if eol_only:
             # Pathspec via stdin: thousands of paths exceed the Windows argv limit.
-            _probe_run(
-                "checkout", "--pathspec-from-file=-", "--pathspec-file-nul", "--",
-                input="\0".join(sorted(eol_only)), check=False,
-            )
-            if _eol_only():
-                # Still dirty: pinning would only surface churn we failed to clear.
+            _probe_run("checkout", "--pathspec-from-file=-", "--pathspec-file-nul", "--",
+                       input="\0".join(sorted(eol_only)), check=False)
+            if _eol_only():  # still dirty: pinning would only surface churn we failed to clear
                 return
             print(f"→ Normalized line-ending churn ({len(eol_only)} file(s))")
-
-        subprocess.run(
-            git_cmd + ["config", "core.autocrlf", "false"], cwd=repo_root, capture_output=True, check=False,
-        )
+        subprocess.run(git_cmd + ["config", "core.autocrlf", "false"], cwd=repo_root, capture_output=True, check=False)