diff --git a/hermes_cli/update_cmd_windows.py b/hermes_cli/update_cmd_windows.py index fe8469c93c..3d076397c1 100644 --- a/hermes_cli/update_cmd_windows.py +++ b/hermes_cli/update_cmd_windows.py @@ -5,7 +5,7 @@ Origin helpers are imported lazily per function (no cycle; test patches on the o """ import logging -from contextlib import suppress +from contextlib import contextmanager, suppress import os import subprocess import sys @@ -19,6 +19,15 @@ from hermes_cli.update_cmd_common import _best_effort logger = logging.getLogger("hermes_cli.update_cmd") +@contextmanager +def _abort_on_error(prefix: str): + """Re-raise any failure of a mandatory step as ``RuntimeError(f"{prefix}: {exc}")`` (chained).""" + try: + yield + except Exception as exc: + raise RuntimeError(f"{prefix}: {exc}") from exc + + def _write_update_planned_stop_marker(profile_path: Path, pid: int) -> bool: """Write a planned-stop marker into a specific profile home.""" try: @@ -28,25 +37,16 @@ def _write_update_planned_stop_marker(profile_path: Path, pid: int) -> bool: 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(), + "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", record, indent=None, separators=(",", ":")) return True except (OSError, PermissionError): return False -def _wait_for_windows_update_gateway_exit( - pids: list[int], *, timeout: float -) -> set[int]: +def _wait_for_windows_update_gateway_exit(pids: list[int], *, timeout: float) -> set[int]: """Wait for the given gateway PIDs to exit, returning survivors.""" if not pids: return set() @@ -83,9 +83,8 @@ def _self_and_non_gateway_ancestor_pids(psutil) -> set[int]: with suppress(Exception): for anc in psutil.Process().parents(): anc_cmdline = _cmdline_or_empty(anc) - if _is_gw is not None and anc_cmdline and _is_gw(anc_cmdline): - continue - skip.add(int(anc.pid)) + if not (_is_gw is not None and anc_cmdline and _is_gw(anc_cmdline)): + skip.add(int(anc.pid)) return skip @@ -121,9 +120,7 @@ def _parent_is_live(proc) -> bool: return parent is not None and parent.is_running() and parent.create_time() <= proc.create_time() -def _detect_venv_python_processes( - *, exclude_pids: set[int] | None = None -) -> list[tuple[int, str, str]]: +def _detect_venv_python_processes(*, exclude_pids: set[int] | None = None) -> list[tuple[int, str, str]]: """Live processes running from the project venv's interpreter as ``(pid, name, cmdline)``; never raises. The hermes.exe shim guard misses the Desktop backend and anything off ``venv\\Scripts\\python(w).exe``; @@ -151,8 +148,7 @@ def _detect_venv_python_processes( info = proc.info except Exception: continue - pid = info.get("pid") - exe = info.get("exe") + pid, exe = info.get("pid"), info.get("exe") if not exe or pid is None or int(pid) in skip: continue try: @@ -163,45 +159,31 @@ def _detect_venv_python_processes( is_holder = 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 is_holder and not (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 and venv_prefix in cmdline_low: - is_holder = True + 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 - if root_prefix in cmdline_low or cwd_low.startswith(root_prefix): - is_holder = True - if not is_holder: - continue - # 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)) + 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 -_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", - } -) - - +_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", +}) _holder_value_flags_cache: frozenset | None = None @@ -242,16 +224,13 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None: except Exception: tokens = cmdline.split() - entry_idx: int | None = None - for i, token in enumerate(tokens): + 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": - entry_idx = i - break - base = low.rsplit("\\", 1)[-1].rsplit("/", 1)[-1] - if base in ("hermes", "hermes.exe"): - entry_idx = i - break + return True + return 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: return None @@ -260,13 +239,11 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None: while i < len(tokens): token = tokens[i] if token in value_flags or token.split("=", 1)[0] in value_flags: - # --flag value consumes two tokens; --flag=value consumes one. - i += 1 if "=" in token else 2 - continue - if token.startswith("-"): + i += 1 if "=" in token else 2 # --flag value consumes two tokens; --flag=value one. + elif token.startswith("-"): i += 1 - continue - return token.lower() + else: + return token.lower() return None @@ -283,17 +260,18 @@ def _format_venv_python_holders_message(matches: list[tuple[int, str, str]]) -> "gateway": " ← gateway", } for pid, name, cmdline in matches[:6]: - sub = _hermes_holder_subcommand(cmdline) - hint = hint_by_subcommand.get(sub or "", "") + hint = hint_by_subcommand.get(_hermes_holder_subcommand(cmdline) or "", "") lines.append(f" PID {pid} {name} {cmdline[:120]}{hint}") if len(matches) > 6: lines.append(f" ... and {len(matches) - 6} more") - lines.append("") - lines.append(" On Windows these keep native extension files (.pyd) locked, so the") - lines.append(" dependency update would fail partway and leave a broken install.") - lines.append(" Close the Hermes desktop app / other Hermes terminals, then re-run:") - lines.append(" hermes update") - lines.append(" (or use `hermes update --force-venv` to proceed anyway at your own risk)") + 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)", + ] return "\n".join(lines) @@ -313,27 +291,19 @@ def _venv_launcher_ancestors(pids: list[int]) -> list[int]: skip = _self_and_non_gateway_ancestor_pids(psutil) found: list[int] = [] for pid in pids: - try: + with suppress(Exception): parent = psutil.Process(int(pid)).parent() - except Exception: - continue - if parent is None: - continue - ppid = int(parent.pid) - if ppid in skip or ppid in found or ppid in set(pids): - continue - try: - exe = (parent.exe() or "").lower() - except Exception: - continue - if exe.startswith(venv_prefix): - found.append(ppid) + 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): + found.append(ppid) return found -def _leftover_pausable_gateway_pids( - matches: list[tuple[int, str, str]], -) -> list[int] | None: +def _leftover_pausable_gateway_pids(matches: list[tuple[int, str, str]]) -> list[int] | None: """PIDs from *matches* when EVERY remaining venv holder is a pausable gateway, else ``None`` (keep refusing). A gateway respawned inside the pause->guard window (or via an unmapped spawn path) still holds ``.pyd`` files. @@ -355,16 +325,13 @@ def _leftover_pausable_gateway_pids( return pids -def _refuse_gateway_ancestor_tree_kill( - pids: list[int], *, gateway_mode: bool -) -> bool: +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. """ if gateway_mode or not pids: return False - try: from hermes_cli.gateway import _is_pid_ancestor_of_current_process @@ -372,18 +339,16 @@ def _refuse_gateway_ancestor_tree_kill( 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})." + f"is running inside it (gateway PID(s): {rendered}).\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." ) - print(" On Windows, taskkill /T would terminate the updater before the update can run.") - print(" From a chat platform, use `/update` instead.") - print(" Otherwise, run `hermes update` from a separate terminal.") return True @@ -429,16 +394,15 @@ def _serve_relaunch_commands(entries: list[dict]) -> list[list[str]]: port = entry.get("port") if not isinstance(port, int) or port <= 0: continue - cmd = [hermes] profile = str(entry.get("profile") or "") - if profile and profile != "default": - cmd += ["--profile", profile] - cmd.append(str(entry.get("purpose"))) host = str(entry.get("host") or "") - if host: - cmd += ["--host", host] - cmd += ["--port", str(port)] - commands.append(cmd) + commands.append( + [hermes] + + (["--profile", profile] if profile and profile != "default" else []) + + [str(entry.get("purpose"))] + + (["--host", host] if host else []) + + ["--port", str(port)] + ) return commands @@ -461,13 +425,10 @@ 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 --port )." - ) + print(" ⚠ Some stopped backends could not be relaunched automatically; " + "restart them manually (hermes serve --host --port ).") _record_update_step( - "serve_relaunch", - not failed and not skipped, + "serve_relaunch", not failed and not skipped, f"relaunched={len(commands) - len(failed)} failed={len(failed)} skipped={skipped}", ) @@ -489,9 +450,7 @@ def _live_argv_low(psutil, pid, cmdline: str) -> str | None: return argv.lower() -def _orphaned_desktop_backend_pids( - matches: list[tuple[int, str, str]], -) -> list[tuple[int, int]] | None: +def _orphaned_desktop_backend_pids(matches: list[tuple[int, str, str]]) -> list[tuple[int, int]] | None: """``(pid, start_time)`` roots from *matches* when every remaining holder is an ORPHANED backend, else ``None``. Killing a Desktop-owned ``serve`` is futile (the app respawns it), but a straggler whose Desktop is gone @@ -546,19 +505,16 @@ def _orphaned_desktop_backend_pids( if not root_set: return None try: - ancestors = {int(a.pid) for a in psutil.Process(pid).parents()} + if not root_set & {int(a.pid) for a in psutil.Process(pid).parents()}: + return None except psutil.NoSuchProcess: continue # exited already except Exception: return None - if not (ancestors & root_set): - return None return roots -def _ledger_reapable_backend_pids( - matches: list[tuple[int, str, str]], -) -> list[int]: +def _ledger_reapable_backend_pids(matches: list[tuple[int, str, str]]) -> list[int]: """PIDs the spawn ledger positively identifies as orphaned backends; never raises. Strongest rung (no PPID/cmdline inference): qualifies when ``(pid, create_time)`` matches a live ledger entry @@ -566,11 +522,7 @@ def _ledger_reapable_backend_pids( 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, - ) + from hermes_cli.process_identity import REAPABLE_PURPOSES, ledger_entries, spawner_is_dead entries = ledger_entries() except Exception: @@ -585,9 +537,7 @@ def _ledger_reapable_backend_pids( ] -def _handoff_reapable_backend_pids( - matches: list[tuple[int, str, str]], -) -> list[int] | None: +def _handoff_reapable_backend_pids(matches: list[tuple[int, str, str]]) -> list[int] | None: """Backend PIDs safe to tree-reap during a GUI-updater hand-off, INCLUDING ones with a live parent; never raises. The orphan-only rung bails on ANY live parent (mid-teardown Electron, launcher->worker chain) and hung a @@ -629,18 +579,12 @@ def _stop_process_trees( if expected_start_time is None: logger.debug("Skipping taskkill of PID %s: process identity unavailable", pid) continue - if not pid_is_hermes( - pid, - expected_start_time=expected_start_time, - ): + if not pid_is_hermes(pid, expected_start_time=expected_start_time): logger.debug("Skipping taskkill of non-Hermes or changed PID %s", pid) continue subprocess.run( - ["taskkill", "/PID", str(pid), "/T", "/F"], - check=False, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - stdin=subprocess.DEVNULL, + ["taskkill", "/PID", str(pid), "/T", "/F"], check=False, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL, creationflags=windows_hide_flags(), ) except Exception as exc: @@ -654,9 +598,7 @@ 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. """ - if "hermes_cli.main" not in (cmdline or "").lower(): - return False - return _hermes_holder_subcommand(cmdline) in ("serve", "dashboard") + return "hermes_cli.main" in (cmdline or "").lower() and _hermes_holder_subcommand(cmdline) in ("serve", "dashboard") def _desktop_owns_gateway_lifecycle() -> bool: @@ -669,11 +611,8 @@ def _desktop_owns_gateway_lifecycle() -> bool: with _best_effort('Desktop-lifecycle ledger probe failed: %s'): from hermes_cli.process_identity import ledger_entries, spawner_is_dead - for entry in ledger_entries(): - if entry.get("purpose") not in ("serve", "dashboard"): - continue - if spawner_is_dead(entry) is False: - return True + if any(e.get("purpose") in ("serve", "dashboard") and spawner_is_dead(e) is False for e in ledger_entries()): + return True psutil = _psutil() try: @@ -696,19 +635,36 @@ def _desktop_owns_gateway_lifecycle() -> bool: def _sc_exe(verb: str, name: str, service, settled_status: str) -> None: """``sc.exe ``; a non-zero exit is only an error when SCM doesn't already report *settled_status*.""" result = subprocess.run( - ["sc.exe", verb, name], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=10, - check=False, + ["sc.exe", verb, name], capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=10, check=False, ) if result.returncode != 0 and service.status() != settled_status: detail = (result.stderr or result.stdout).strip() raise RuntimeError(detail or f"sc.exe {verb} failed with {result.returncode}") +def _original_process_is_alive(psutil, pid: int, create_time: float) -> bool: + """Is the process with this exact ``(pid, create_time)`` identity still alive? AccessDenied/unknown reads + True (fail closed: the venv may still be locked).""" + try: + current = float(psutil.Process(pid).create_time()) + except (psutil.NoSuchProcess, psutil.ZombieProcess): + return False + except Exception: + return True + return abs(current - create_time) <= 0.001 + + +def _poll_until(condition, timeout: float, interval: float = 0.2) -> bool: + """Poll *condition* every *interval* seconds until true (True) or *timeout* elapses (False).""" + deadline = _time.monotonic() + timeout + while _time.monotonic() < deadline: + if condition(): + return True + _time.sleep(interval) + return False + + def _process_create_time(psutil, pid: int, label: str, when: str) -> float: try: return float(psutil.Process(int(pid)).create_time()) @@ -716,33 +672,23 @@ def _process_create_time(psutil, pid: int, label: str, when: str) -> float: raise RuntimeError(f"Windows {label} process identity is unavailable {when}") from exc -def _verify_service_identities( - psutil, name: str, service, expected_service_identity, expected_gateway_identity -) -> None: +def _verify_service_identities(psutil, name: str, service, expected_service_identity, expected_gateway_identity) -> None: """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) except Exception as exc: - raise RuntimeError( - f"Windows service {name} SCM identity is unavailable before stop" - ) from exc + raise RuntimeError(f"Windows service {name} SCM identity is unavailable before stop") from exc if current_status != "running": - raise RuntimeError( - f"Windows service {name} is not stably running before stop: {current_status}" - ) + raise RuntimeError(f"Windows service {name} is not stably running before stop: {current_status}") 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), - ): + for label, identity in (("service", expected_service_identity), ("gateway", expected_gateway_identity)): if identity is None: continue pid, create_time = identity - current = _process_create_time(psutil, pid, label, "before stop") - if abs(current - float(create_time)) > 0.001: + if abs(_process_create_time(psutil, pid, label, "before stop") - float(create_time)) > 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]) @@ -750,16 +696,13 @@ def _verify_service_identities( try: ancestor_pids = {int(parent.pid) for parent in psutil.Process(gateway_pid).parents()} except Exception as exc: - raise RuntimeError( - "Windows gateway ancestry is unavailable before service stop" - ) from exc + raise RuntimeError("Windows gateway ancestry is unavailable before service stop") from exc if service_pid not in ancestor_pids: raise RuntimeError(f"Windows gateway is no longer owned by service {name}") def _stop_windows_gateway_service( - name: str, - *, + name: str, *, expected_processes: tuple[tuple[int, float], ...] = (), expected_service_identity: tuple[int, float] | None = None, expected_gateway_identity: tuple[int, float] | None = None, @@ -772,35 +715,16 @@ def _stop_windows_gateway_service( _verify_service_identities(psutil, name, service, expected_service_identity, expected_gateway_identity) _sc_exe("stop", name, service, "stopped") - def _original_process_is_alive(pid: int, create_time: float) -> bool: - try: - current = float(psutil.Process(pid).create_time()) - except (psutil.NoSuchProcess, psutil.ZombieProcess): - return False - except Exception: - return True # AccessDenied/unknown: fail closed, venv may still be locked - return abs(current - create_time) <= 0.001 - def _alive() -> list[int]: - return [pid for pid, create_time in expected_processes if _original_process_is_alive(pid, create_time)] + return [pid for pid, create_time in expected_processes if _original_process_is_alive(psutil, pid, create_time)] - deadline = _time.monotonic() + timeout - while _time.monotonic() < deadline: - if service.status() == "stopped" and not _alive(): - return - _time.sleep(0.2) - if service.status() == "stopped": + 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: - raise RuntimeError( - f"Windows service {name} stopped but its process tree is still alive: " - f"{alive_after_stop}" - ) - return - raise RuntimeError( - f"Windows service {name} did not stop within {timeout:.0f}s; venv mutation unsafe." - ) + 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: @@ -809,12 +733,8 @@ def _start_windows_gateway_service(name: str, *, timeout: float = 30.0) -> None: service = psutil.win_service_get(name) _sc_exe("start", name, service, "running") - deadline = _time.monotonic() + timeout - while _time.monotonic() < deadline: - if service.status() == "running": - return - _time.sleep(0.2) - raise RuntimeError(f"Windows service {name} did not start within {timeout:.0f}s") + if not _poll_until(lambda: service.status() == "running", timeout): + raise RuntimeError(f"Windows service {name} did not start within {timeout:.0f}s") def _restore_windows_gateway_service(name: str, *, timeout: float = 60.0) -> None: @@ -832,9 +752,7 @@ def _restore_windows_gateway_service(name: str, *, timeout: float = 60.0) -> Non _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" - ) + raise RuntimeError(f"Windows service {name} did not reach a restorable state within {timeout:.0f}s") def _windows_cold_start_plan() -> dict | None: @@ -849,22 +767,13 @@ def _windows_cold_start_plan() -> dict | None: 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" - ) + 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 {"resume_needed": True, "profiles": {}, "unmapped_pids": [], "unmapped": [], "cold_start_if_installed": True} return None @@ -883,17 +792,9 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic current_service_name = str(service.name) _stop_windows_gateway_service( 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), - ), + 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)), ) paused_services.append(current_service_name) current_service_name = None @@ -909,23 +810,17 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic print(" ✓ Paused Windows gateway service(s): " + ", ".join(paused_services)) return token except Exception as exc: - restore_names = [] - if current_service_name: - restore_names.append(current_service_name) - restore_names.extend(reversed(paused_services)) - rollback_failures = [] - for service_name in dict.fromkeys(restore_names): - try: - _restore_windows_gateway_service(service_name) - except Exception as restore_exc: - rollback_failures.append(f"{service_name}: {restore_exc}") + restore_names = ([current_service_name] if current_service_name else []) + list(reversed(paused_services)) + rollback: list[tuple[str, object]] = [(n, lambda n=n: _restore_windows_gateway_service(n)) for n in dict.fromkeys(restore_names)] if profiles or unmapped: + rollback.append(("ordinary gateways", lambda: _resume_windows_gateways_after_update(token))) + rollback_failures = [] + for label, restore in rollback: try: - _resume_windows_gateways_after_update(token) + restore() except Exception as restore_exc: - rollback_failures.append(f"ordinary gateways: {restore_exc}") - failed_service = current_service_name or "unknown" - detail = f"Could not stop Windows gateway service {failed_service}: {exc}" + rollback_failures.append(f"{label}: {restore_exc}") + detail = f"Could not stop Windows gateway service {current_service_name or 'unknown'}: {exc}" if rollback_failures: detail += "; rollback failures: " + "; ".join(rollback_failures) raise RuntimeError(detail) from exc @@ -933,36 +828,18 @@ 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.""" - from hermes_cli.gateway import ( - find_gateway_pids, - find_profile_gateway_processes, - find_windows_gateway_services, - ) + from hermes_cli.gateway import find_gateway_pids, find_profile_gateway_processes, find_windows_gateway_services - try: + 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} - except Exception as exc: - raise RuntimeError(f"Could not map Windows gateway PIDs to profiles: {exc}") from exc - - try: + with _abort_on_error("Could not determine Windows gateway service ownership"): service_gateways = find_windows_gateway_services(profile_processes=profile_process_list) - except Exception as exc: - raise RuntimeError(f"Could not determine Windows gateway service ownership: {exc}") from exc - service_gateway_pids = {int(service.gateway_pid) for service in service_gateways} - try: - running_pids = list( - dict.fromkeys( - [ - *find_gateway_pids(all_profiles=True), - *sorted(profile_processes), - *sorted(service_gateway_pids), - ] - ) - ) - except Exception as exc: - raise RuntimeError(f"Could not discover Windows gateway PIDs before update: {exc}") from exc + with _abort_on_error("Could not discover Windows gateway PIDs before update"): + running_pids = list(dict.fromkeys( + [*find_gateway_pids(all_profiles=True), *sorted(profile_processes), *sorted(service_gateway_pids)] + )) return profile_processes, service_gateways, service_gateway_pids, running_pids @@ -976,9 +853,7 @@ def _request_socket_pauses(running_pids, profile_processes, service_gateway_pids mapped_pids = [] socket_acks: list[dict] = [] for pid in running_pids: - if pid in service_gateway_pids: - continue - proc = profile_processes.get(pid) + proc = None if pid in service_gateway_pids else profile_processes.get(pid) if proc is None: continue profiles[str(proc.profile)] = int(pid) @@ -1007,10 +882,7 @@ def _gateway_drain_timeout(socket_acks: list[dict]) -> float: with suppress(Exception): declared = max(float(a.get("drain_timeout") or 0.0) for a in socket_acks) drain_timeout = max(drain_timeout, declared + 10.0) - print( - f" → {len(socket_acks)} gateway(s) ACKed socket pause; " - f"waiting up to {int(drain_timeout)}s for graceful exit" - ) + print(f" → {len(socket_acks)} gateway(s) ACKed socket pause; waiting up to {int(drain_timeout)}s for graceful exit") return drain_timeout @@ -1025,19 +897,15 @@ def _pause_windows_gateways_for_update() -> dict | None: if not _m()._is_windows(): return None - try: + 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 - except Exception as exc: - raise RuntimeError(f"Could not prepare Windows gateway pause for update: {exc}") from exc 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 - ) + 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 @@ -1047,11 +915,7 @@ def _pause_windows_gateways_for_update() -> dict | None: 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) - unmapped_pids = [ - pid - for pid in running_pids - if pid not in profile_processes and pid not in service_gateway_pids - ] + unmapped_pids = [pid for pid in running_pids if pid not in profile_processes and pid not in service_gateway_pids] # 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 ...``). @@ -1084,13 +948,7 @@ def _pause_windows_gateways_for_update() -> dict | None: # No recoverable cmdline (psutil missing, access denied, gone): manual restart. print(" Restart manually after update: hermes gateway run") - token = { - "resume_needed": True, - "profiles": profiles, - "unmapped_pids": unmapped_pids, - "unmapped": unmapped, - } - + token = {"resume_needed": True, "profiles": profiles, "unmapped_pids": unmapped_pids, "unmapped": unmapped} return _pause_windows_gateway_services(service_gateways, token, profiles, unmapped) @@ -1103,43 +961,24 @@ def _cold_start_windows_gateway_after_update() -> bool: from hermes_cli.update_cmd import _desktop_owns_gateway_lifecycle, _m if not _m()._is_windows(): return True - try: + with _abort_on_error("Could not load Windows gateway cold-start helpers"): from hermes_cli import gateway_windows from hermes_cli.gateway import find_gateway_pids - except Exception as exc: - raise RuntimeError(f"Could not load Windows gateway cold-start helpers: {exc}") from exc - - try: + with _abort_on_error("Could not re-check gateway liveness before cold-start"): if list(find_gateway_pids(all_profiles=True)): return True - except Exception as exc: - raise RuntimeError(f"Could not re-check gateway liveness before cold-start: {exc}") from exc - - try: + with _abort_on_error("Could not re-check Desktop gateway-lifecycle ownership before cold-start"): if _desktop_owns_gateway_lifecycle(): logger.debug("Skipping Windows gateway cold-start: Desktop owns gateway lifecycle") return True - except Exception as exc: - raise RuntimeError( - "Could not re-check Desktop gateway-lifecycle ownership before cold-start: " - f"{exc}" - ) from exc - - try: + with _abort_on_error("Could not cold-start Windows gateway after update"): pid = gateway_windows._spawn_detached() - except Exception as exc: - raise RuntimeError(f"Could not cold-start Windows gateway after update: {exc}") from exc - if not pid: raise RuntimeError("Windows gateway cold-start did not return a process ID") ready_pids = gateway_windows._wait_for_gateway_ready() if not ready_pids: raise RuntimeError(f"Windows gateway cold-start PID {pid} did not become ready") - print() - print( - "✓ Gateway started via cold-start after update " - f"(PID: {', '.join(map(str, ready_pids))})" - ) + 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") @@ -1158,10 +997,9 @@ def _refresh_windows_gateway_launchers() -> None: with _best_effort('Could not refresh Windows gateway launchers after update: %s'): from hermes_cli import gateway_windows - if not gateway_windows.is_installed(): - return - gateway_windows._write_task_script() - print(" ✓ Refreshed Windows gateway launcher scripts") + if gateway_windows.is_installed(): + gateway_windows._write_task_script() + print(" ✓ Refreshed Windows gateway launcher scripts") def _refresh_bootstrap_cache_scripts(branch: str = "main") -> None: @@ -1186,10 +1024,8 @@ def _refresh_bootstrap_cache_scripts(branch: str = "main") -> None: refreshed = [] for kind, src_name in (("ps1", "install.ps1"), ("sh", "install.sh")): src = _m().PROJECT_ROOT / "scripts" / src_name - if not src.is_file(): - continue cached = cache_dir / f"install-{safe_ref}.{kind}" - if not cached.is_file(): + if not src.is_file() or not cached.is_file(): 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"): @@ -1202,10 +1038,7 @@ def _refresh_bootstrap_cache_scripts(branch: str = "main") -> None: os.replace(tmp, cached) refreshed.append(cached.name) if refreshed: - print( - " ✓ Refreshed installer bootstrap-cache script(s): " - + ", ".join(sorted(refreshed)) - ) + print(" ✓ Refreshed installer bootstrap-cache script(s): " + ", ".join(sorted(refreshed))) def _resume_windows_services(token: dict) -> None: @@ -1223,25 +1056,17 @@ def _resume_windows_services(token: dict) -> None: if str(service_name) not in verified_restarts: verified_restarts.append(str(service_name)) except Exception as exc: - logger.warning( - "Could not restart Windows gateway service %s after update: %s", - service_name, - 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)) token["restarted_services"] = verified_restarts if failed_services: token["services"] = failed_services - raise RuntimeError( - "Could not restart Windows gateway service(s): " - + ", ".join(failed_services) - ) + raise RuntimeError("Could not restart Windows gateway service(s): " + ", ".join(failed_services)) token["services"] = [] if restarted_services: - print() - print(" ✓ Restarted Windows gateway service(s): " + ", ".join(restarted_services)) + print("\n ✓ Restarted Windows gateway service(s): " + ", ".join(restarted_services)) def _relaunch_paused_gateways(token: dict, profiles: dict, unmapped: list) -> tuple[list[str], int]: @@ -1250,50 +1075,39 @@ def _relaunch_paused_gateways(token: dict, profiles: dict, unmapped: list) -> tu 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. """ - try: - from hermes_cli.gateway import ( - launch_detached_gateway_restart_by_cmdline, - launch_detached_profile_gateway_restart, - ) - except Exception as exc: - raise RuntimeError(f"Could not load Windows gateway restart helper: {exc}") from exc + 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 relaunched = [] failed_profiles = {} for profile, old_pid in sorted(profiles.items()): - try: - if launch_detached_profile_gateway_restart(str(profile), int(old_pid)): - relaunched.append(str(profile)) - else: - failed_profiles[str(profile)] = int(old_pid) - except Exception as exc: - logger.debug( - "Could not restart Windows gateway profile %s after update: %s", - profile, - exc, - ) + 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 = entry.get("argv") - old_pid = entry.get("pid") - if not argv or not old_pid: - failed_unmapped.append(entry) - continue - try: - if launch_detached_gateway_restart_by_cmdline(int(old_pid), list(argv)): - unmapped_relaunched += 1 - else: - failed_unmapped.append(entry) - except Exception as exc: - logger.debug( - "Could not restart unmapped Windows gateway (pid %s) after update: %s", - old_pid, - exc, - ) + 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, + ): + unmapped_relaunched += 1 + else: failed_unmapped.append(entry) token["profiles"] = failed_profiles @@ -1310,24 +1124,19 @@ def _verify_relaunched_gateways_alive(token: dict, profiles: dict, unmapped: lis ``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). """ - try: + with _abort_on_error("Could not load Windows gateway liveness helpers"): from hermes_cli import gateway_windows - except Exception as exc: - raise RuntimeError(f"Could not load Windows gateway liveness helpers: {exc}") from exc ready_pids = gateway_windows._wait_for_gateway_ready(timeout_s=30.0, all_profiles=True) if not ready_pids: token["profiles"] = dict(profiles) token["unmapped"] = list(unmapped) - print() - print( - " ⚠ Windows gateway restart could not be verified — no stable " - "gateway process appeared after relaunch." - ) 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.)" + "Job Object during updater teardown, #48820.)\n" + " Recover with: hermes gateway restart" ) - print(" Recover with: hermes gateway restart") raise RuntimeError("Windows gateway relaunch after update was not verified alive") with suppress(Exception): gateway_windows._write_start_attestation(ready_pids, "post-update relaunch") @@ -1363,12 +1172,10 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None: token["resume_needed"] = False if relaunched: - print() - print(f" ✓ Restarting Windows gateway profile(s): {', '.join(relaunched)}") + print(f"\n ✓ Restarting Windows gateway profile(s): {', '.join(relaunched)}") if unmapped_relaunched: - if not relaunched: - print() - print(f" ✓ Restarting {unmapped_relaunched} unmapped Windows gateway process(es)") + lead = "" if relaunched else "\n" + print(f"{lead} ✓ Restarting {unmapped_relaunched} unmapped Windows gateway process(es)") def _resume_windows_gateways_and_merge_outcome(outcome, _windows_gateway_resume, gateway_mode: bool): @@ -1387,24 +1194,17 @@ def _resume_windows_gateways_and_merge_outcome(outcome, _windows_gateway_resume, if isinstance(_windows_gateway_resume, dict): def _extend_unique(target: list, items) -> None: - for item in items: - if item not in target: - target.append(item) + 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, _windows_gateway_resume.get("relaunched_profiles") or []) - windows_restarted = list(_windows_gateway_resume.get("restarted_services") or []) - service_profiles = _windows_gateway_resume.get("service_profiles") or {} + _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 (_windows_gateway_resume.get("services") or [])), - ) + _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 @@ -1415,10 +1215,7 @@ def _resume_windows_gateways_and_merge_outcome(outcome, _windows_gateway_resume, 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) - ), + incomplete=outcome.incomplete or bool(outcome.failed_or_stale_units), phase_error="; ".join(outcome.phase_errors) or None, ) @@ -1438,8 +1235,7 @@ def _terminate_leftover_gateways(pids) -> None: for _pid in pids: try: - pid_int = int(_pid) - terminate_pid(pid_int, force=True, expected_start_time=get_process_start_time(pid_int)) + 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) @@ -1451,12 +1247,8 @@ def _in_handoff_without_live_shim(args) -> bool: """ from hermes_cli.update_cmd import _m try: - handoff = bool(getattr(args, "gateway", False)) and _m()._update_marker_path().exists() - except Exception: - return False - if not handoff: - return False - try: + if not (bool(getattr(args, "gateway", False)) and _m()._update_marker_path().exists()): + return False scripts_dir = _m()._venv_scripts_dir() return scripts_dir is not None and not _m()._detect_concurrent_hermes_instances(scripts_dir) except Exception: @@ -1481,10 +1273,8 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew _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, + f" ⚠ {len(_gateway_holders)} gateway process(es) still hold the venv after the pause; stopping them", + _gateway_holders, stop=_terminate_leftover_gateways, ) if _venv_holders: # Positive-identity rung (any context): spawn ledger proves the holder is an @@ -1493,8 +1283,7 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew if _ledger_backends: _venv_holders = _reap_and_rescan( f" ⚠ {len(_ledger_backends)} ledger-identified orphaned " - "Hermes backend process(es) hold the venv; stopping their trees", - _ledger_backends, + "Hermes backend process(es) hold the venv; stopping their trees", _ledger_backends, ) if _venv_holders: # Desktop `serve` backends whose app is GONE: nothing respawns an orphan, so @@ -1503,8 +1292,7 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew if _orphan_backends: _venv_holders = _reap_and_rescan( f" ⚠ {len(_orphan_backends)} orphaned Desktop backend " - "process(es) still hold the venv; stopping their trees", - _orphan_backends, + "process(es) still hold the venv; stopping their trees", _orphan_backends, ) if _venv_holders: # Manual serve/dashboard rung (e.g. `hermes serve --host ` for a REMOTE Desktop): @@ -1517,16 +1305,12 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew _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, + 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` @@ -1536,9 +1320,7 @@ def _clear_windows_venv_holders_or_exit(args, gateway_mode: bool, _windows_gatew 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, + "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)) diff --git a/hermes_cli/update_restart_recovery.py b/hermes_cli/update_restart_recovery.py index 3b92865c69..547b7bb06e 100644 --- a/hermes_cli/update_restart_recovery.py +++ b/hermes_cli/update_restart_recovery.py @@ -70,24 +70,12 @@ _UNIT_SETTLE_ATTEMPTS = 10 _UNIT_SETTLE_DELAY = 1.0 -def _run_quiet( - run: Callable[..., Any], - argv: list[str], - *, - timeout: int, - **extra: Any, -) -> Any | None: +def _run_quiet(run: Callable[..., Any], argv: list[str], *, timeout: int, **extra: Any) -> Any | None: """Run ``argv`` capturing text output; ``None`` when it errors or times out.""" try: return run( - argv, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - check=False, - timeout=timeout, - **extra, + argv, capture_output=True, text=True, encoding="utf-8", errors="replace", + check=False, timeout=timeout, **extra, ) except (OSError, subprocess.TimeoutExpired): return None @@ -114,10 +102,7 @@ def _run_profile_restart(profile: str, *, run: Callable[..., Any]) -> bool: """Run one profile restart without inheriting the updater's process state.""" kwargs: dict[str, Any] = {"stdin": subprocess.DEVNULL, "env": _child_environment()} if os.name == "nt": - kwargs["creationflags"] = ( - getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) - | getattr(subprocess, "DETACHED_PROCESS", 0) - ) + kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr(subprocess, "DETACHED_PROCESS", 0) else: kwargs["start_new_session"] = True argv = [sys.executable, "-m", "hermes_cli.main", "-p", profile, "gateway", "restart"] @@ -132,26 +117,18 @@ def _systemd_unit_candidates(profile: str) -> tuple[str, ...]: def _systemd_verified_active(profile: str, *, run: Callable[..., Any]) -> bool: - """Return True only when systemd itself reports the profile's unit active. - - This is the observation that separates ``verified`` from ``relaunch_attempted``. Any failure - here (no ``systemctl``, probe error, unit not ``active``) means we could NOT verify — never that - the restart failed. - """ + """True only when systemd itself reports the profile's unit active — the observation separating + ``verified`` from ``relaunch_attempted``. Any failure (no ``systemctl``, probe error, unit not + ``active``) means we could NOT verify, never that the restart failed.""" systemctl = shutil.which("systemctl") - if not systemctl: - return False - return any( + return bool(systemctl) and any( _unit_is_active([systemctl, "--user"], unit, run=run, require_rc0=True) for unit in _systemd_unit_candidates(profile) ) def restart_profiles( - profiles: Iterable[str], - *, - supervisors: Mapping[str, str] | None = None, - run: Callable[..., Any] = subprocess.run, + profiles: Iterable[str], *, supervisors: Mapping[str, str] | None = None, run: Callable[..., Any] = subprocess.run ) -> dict[str, list[str]]: """Restart the supplied profiles (only ones whose inventory identified a service supervisor). @@ -172,15 +149,11 @@ def restart_profiles( def _systemctl_scopes() -> list[tuple[str, list[str]]]: - """``systemctl`` invocations for the user and system scopes, or nothing. + """``(label, systemctl argv)`` for the user and system scopes (the pair the in-process phase walks), or nothing. - Mirrors the scope pair the in-process restart phase walks. ``systemctl`` is resolved through - ``shutil.which`` so this module never has to import any Hermes platform helper — importing the - freshly pulled tree is exactly what aborted the phase that called us. - - Each scope is returned with its label because ``hermes-serve.service`` in the user manager and - ``hermes-serve.service`` in the system manager are two different processes. Every identity this - module produces or consumes stays qualified by that label; the bare unit name is never the key. + ``systemctl`` comes from ``shutil.which`` so this module never imports a Hermes platform helper — + importing the freshly pulled tree is exactly what aborted the phase that called us. Scopes carry + their label because the same unit name in both managers is two different processes. """ systemctl = shutil.which("systemctl") if not systemctl or sys.platform != "linux": @@ -209,18 +182,14 @@ def _listed_serve_units(scope: list[str], *, run: Callable[..., Any]) -> list[st def _unit_main_pid(scope: list[str], unit: str, *, run: Callable[..., Any]) -> int: """The unit's ``MainPID`` via ``systemctl show``; ``0`` when absent or unreadable.""" - result = _run_quiet( - run, scope + ["show", unit, "--property=MainPID", "--value"], timeout=_VERIFY_TIMEOUT - ) + result = _run_quiet(run, scope + ["show", unit, "--property=MainPID", "--value"], timeout=_VERIFY_TIMEOUT) try: return int(_stdout(result) or 0) if _succeeded(result) else 0 except ValueError: return 0 -def _unit_is_active( - scope: list[str], unit: str, *, run: Callable[..., Any], require_rc0: bool = False -) -> bool: +def _unit_is_active(scope: list[str], unit: str, *, run: Callable[..., Any], require_rc0: bool = False) -> bool: result = _run_quiet(run, scope + ["is-active", unit], timeout=_VERIFY_TIMEOUT) if result is None or (require_rc0 and not _succeeded(result)): return False @@ -228,12 +197,7 @@ def _unit_is_active( def _serve_unit_replaced( - scope: list[str], - unit: str, - previous_pid: int, - *, - run: Callable[..., Any], - sleep: Callable[[float], Any], + scope: list[str], unit: str, previous_pid: int, *, run: Callable[..., Any], sleep: Callable[[float], Any] ) -> bool: """Did the unit come back on a NEW main process? @@ -244,11 +208,10 @@ def _serve_unit_replaced( for attempt in range(_UNIT_SETTLE_ATTEMPTS): if attempt: sleep(_UNIT_SETTLE_DELAY) - if not _unit_is_active(scope, unit, run=run): - continue - current = _unit_main_pid(scope, unit, run=run) - if current > 0 and current != previous_pid: - return True + if _unit_is_active(scope, unit, run=run): + current = _unit_main_pid(scope, unit, run=run) + if current > 0 and current != previous_pid: + return True return False @@ -264,38 +227,25 @@ def _split_skip_entry(entry: Any) -> tuple[str, str]: return scope_label, unit.removesuffix(".service") -def _normalized_skips( - skip_units: Iterable[Any], -) -> tuple[set[tuple[str, str]], set[str]]: - """Split already-settled units into scope-qualified and legacy entries. - - A bare ``"hermes-serve"`` carries no scope and therefore cannot say WHICH of two same-named - processes was settled. - """ +def _normalized_skips(skip_units: Iterable[Any]) -> tuple[set[tuple[str, str]], set[str]]: + """Split already-settled units into scope-qualified and legacy entries (a bare ``"hermes-serve"`` + carries no scope and cannot say WHICH of two same-named processes was settled).""" qualified: set[tuple[str, str]] = set() legacy: set[str] = set() for entry in skip_units or (): scope_label, base = _split_skip_entry(entry) - if not base: - continue - if scope_label in _SCOPE_LABELS: - qualified.add((scope_label, base)) - else: - legacy.add(base) + if base: + (qualified.add((scope_label, base)) if scope_label in _SCOPE_LABELS else legacy.add(base)) return qualified, legacy def restart_serve_units( - *, - skip_units: Iterable[Any] = (), - run: Callable[..., Any] = subprocess.run, - sleep: Callable[[float], Any] = time.sleep, + *, skip_units: Iterable[Any] = (), run: Callable[..., Any] = subprocess.run, sleep: Callable[[float], Any] = time.sleep ) -> dict[str, list[str]]: """Restart every active ``hermes-serve*`` systemd unit from this process. - Units are enumerated from systemd, never from the update inventory. That keeps the relaunch - authority requirement structural: a manually launched or Desktop-owned ``hermes serve`` owns no - unit and therefore cannot be touched here. + Units are enumerated from systemd, never from the update inventory, so a manually launched or + Desktop-owned ``hermes serve`` (no unit) structurally cannot be touched here. """ skipped_qualified, skipped_legacy = _normalized_skips(skip_units) # (scope, base unit) -> replaced? The same unit name in the user and the system @@ -333,10 +283,7 @@ def _parse_payload(stream) -> tuple[list[str], dict[str, str], bool, list[str]]: profiles = payload.get("profiles") if not isinstance(profiles, list): raise ValueError("recovery payload must contain a profiles list") - if any( - not isinstance(profile, str) or not _PROFILE_ID_RE.fullmatch(profile) - for profile in profiles - ): + if any(not isinstance(profile, str) or not _PROFILE_ID_RE.fullmatch(profile) for profile in profiles): raise ValueError("recovery profiles contain an invalid profile id") raw_supervisors = payload.get("supervisors") supervisors: dict[str, str] = {} @@ -361,27 +308,20 @@ def _parse_payload(stream) -> tuple[list[str], dict[str, str], bool, list[str]]: if not isinstance(raw_skip, list) or any(not isinstance(entry, (str, dict)) for entry in raw_skip): raise ValueError("recovery serve_units skip list is invalid") for entry in raw_skip: - # A skip entry names one already-settled process. The qualified - # shape carries the systemd scope, because `hermes-serve.service` - # can exist in BOTH managers and settling one says nothing about - # the other. A bare string is the legacy shape a pre-update - # interpreter can still send; it is kept, and read as - # scope-agnostic by `restart_serve_units`. + # A skip entry names one already-settled process. The qualified shape carries the systemd + # scope (`hermes-serve.service` can exist in BOTH managers; settling one says nothing about + # the other). A bare string is the legacy shape a pre-update interpreter can still send, + # read as scope-agnostic by `restart_serve_units`. if isinstance(entry, dict): if not isinstance(entry.get("unit"), str): raise ValueError("recovery serve_units skip list is invalid") if not isinstance(entry.get("scope"), str): entry = {"unit": entry["unit"]} scope_label, base = _split_skip_entry(entry) - # Only the shapes systemd can actually produce for this family; a - # skip entry is a name filter, never a command argument. An - # unrecognized scope drops the entry rather than raising: dropping - # a skip can only ever cause one more restart-and-verify, while - # honouring an unreadable one could suppress recovery of a stale - # process. - if scope_label and scope_label not in _SCOPE_LABELS: - continue - if not _UNIT_RE.fullmatch(f"{base}.service"): + # Only shapes systemd can produce for this family (a skip is a name filter, never a command + # argument). An unrecognized scope DROPS the entry rather than raising: a dropped skip costs at + # most one extra restart-and-verify; honouring an unreadable one could suppress recovery. + if (scope_label and scope_label not in _SCOPE_LABELS) or not _UNIT_RE.fullmatch(f"{base}.service"): continue skip_units.append(f"{scope_label}/{base}" if scope_label else base) return profiles, supervisors, recover_serve, skip_units @@ -402,10 +342,7 @@ def main(argv: list[str] | None = None) -> int: ) except (ValueError, json.JSONDecodeError) as exc: print(json.dumps({ - "error": str(exc), - "verified": [], - "relaunch_attempted": [], - "failed": [], + "error": str(exc), "verified": [], "relaunch_attempted": [], "failed": [], "serve_units": {"verified": [], "failed": []}, })) return 2