diff --git a/gateway/run.py b/gateway/run.py index 24d501b5b752..813ebff1b28d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1702,6 +1702,31 @@ def _clear_planned_restart_notification() -> None: from utils import atomic_json_write, is_truthy_value _hermes_home = get_hermes_home() + +def _current_systemd_gateway_service() -> str | None: + """Return the exact gateway service component containing this process.""" + from hermes_cli.update_owner_restart import current_systemd_gateway_service + + return current_systemd_gateway_service() + + +def _read_update_terminal_exit_code(home: Path) -> int | None: + """Read a generic exit marker or the atomic owner-restart result.""" + exit_code_path = Path(home) / ".update_exit_code" + if exit_code_path.exists(): + try: + return int(exit_code_path.read_text(encoding="utf-8").strip() or "1") + except (OSError, TypeError, ValueError): + return 1 + try: + from hermes_cli.update_owner_restart import ( + read_owner_restart_result_exit_code, + ) + + return read_owner_restart_result_exit_code(Path(home)) + except Exception: + return None + # Load environment variables from ~/.hermes/.env first. # User-managed env files should override stale shell exports on restart. from dotenv import load_dotenv # noqa: F401 # backward-compat for tests that monkeypatch this symbol @@ -11271,6 +11296,25 @@ async def start(self) -> bool: logger.info("Channel directory built: %d target(s)", ch_count) except Exception as e: logger.warning("Channel directory build failed: %s", e) + + # A transient verifier in a separate systemd cgroup owns the terminal + # update result. Acknowledge only after this new owner has completed + # adapter/config startup and published its running runtime state. The + # verifier independently checks ActiveState, MainPID, and start + # generation before it exposes .update_exit_code. + try: + from hermes_cli.update_owner_restart import ( + acknowledge_owner_restart_ready, + ) + + if acknowledge_owner_restart_ready( + _hermes_home, + current_service=_current_systemd_gateway_service(), + current_pid=os.getpid(), + ): + logger.info("Acknowledged updater-owner restart readiness") + except Exception as exc: + logger.warning("Could not acknowledge updater-owner readiness: %s", exc) # Check if we're restarting after a /update command. If the update is # still running, keep watching so we notify once it actually finishes. @@ -20565,6 +20609,7 @@ async def _watch_update_progress( claimed_path = _hermes_home / ".update_pending.claimed.json" output_path = _hermes_home / ".update_output.txt" exit_code_path = _hermes_home / ".update_exit_code" + owner_result_path = _hermes_home / ".update_owner_restart_result.json" prompt_path = _hermes_home / ".update_prompt.json" loop = asyncio.get_running_loop() @@ -20613,10 +20658,15 @@ async def _watch_update_progress( # after the first completion check — otherwise a platform that # reconnects a few seconds after completion never gets notified. while (pending_path.exists() or claimed_path.exists()) and loop.time() < deadline: - if exit_code_path.exists() and await self._send_update_notification(): + if ( + _read_update_terminal_exit_code(_hermes_home) is not None + and await self._send_update_notification() + ): return await asyncio.sleep(poll_interval) - if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists(): + if ( + pending_path.exists() or claimed_path.exists() + ) and _read_update_terminal_exit_code(_hermes_home) is None: exit_code_path.write_text("124", encoding="utf-8") await self._send_update_notification() return @@ -20655,8 +20705,10 @@ async def _flush_buffer() -> None: logger.debug("Update stream send failed: %s", e) while loop.time() < deadline: - # Check for completion - if exit_code_path.exists(): + # Check for completion. Owner-managed updates commit their atomic + # result JSON before best-effort projection to .update_exit_code. + terminal_exit_code = _read_update_terminal_exit_code(_hermes_home) + if terminal_exit_code is not None: # Read any remaining output if output_path.exists(): try: @@ -20670,8 +20722,7 @@ async def _flush_buffer() -> None: # Send final status try: - exit_code_raw = exit_code_path.read_text(encoding="utf-8").strip() or "1" - exit_code = int(exit_code_raw) + exit_code = terminal_exit_code if exit_code == 0: await adapter.send( chat_id, @@ -20689,8 +20740,14 @@ async def _flush_buffer() -> None: logger.warning("Update final notification failed: %s", e) # Cleanup - for p in (pending_path, claimed_path, output_path, - exit_code_path, prompt_path): + for p in ( + pending_path, + claimed_path, + output_path, + exit_code_path, + owner_result_path, + prompt_path, + ): p.unlink(missing_ok=True) (_hermes_home / ".update_response").unlink(missing_ok=True) _up_done = self._peek_session_state(session_key) @@ -20773,7 +20830,7 @@ async def _flush_buffer() -> None: await asyncio.sleep(poll_interval) # Timeout - if not exit_code_path.exists(): + if _read_update_terminal_exit_code(_hermes_home) is None: logger.warning("Update watcher timed out after %.0fs", timeout) exit_code_path.write_text("124", encoding="utf-8") await _flush_buffer() @@ -20785,8 +20842,14 @@ async def _flush_buffer() -> None: ) except Exception: pass - for p in (pending_path, claimed_path, output_path, - exit_code_path, prompt_path): + for p in ( + pending_path, + claimed_path, + output_path, + exit_code_path, + owner_result_path, + prompt_path, + ): p.unlink(missing_ok=True) (_hermes_home / ".update_response").unlink(missing_ok=True) _up_timeout_state = self._peek_session_state(session_key) @@ -20807,6 +20870,7 @@ async def _send_update_notification(self) -> bool: claimed_path = _hermes_home / ".update_pending.claimed.json" output_path = _hermes_home / ".update_output.txt" exit_code_path = _hermes_home / ".update_exit_code" + owner_result_path = _hermes_home / ".update_owner_restart_result.json" if not pending_path.exists() and not claimed_path.exists(): return False @@ -20830,16 +20894,14 @@ async def _send_update_notification(self) -> bool: thread_id = pending.get("thread_id") message_id = pending.get("message_id") - if not exit_code_path.exists(): + exit_code = _read_update_terminal_exit_code(_hermes_home) + if exit_code is None: logger.info("Update notification deferred: update still running") cleanup = False active_pending_path = pending_path claimed_path.replace(pending_path) return False - exit_code_raw = exit_code_path.read_text(encoding="utf-8").strip() or "1" - exit_code = int(exit_code_raw) - # Read the captured update output output = "" if output_path.exists(): @@ -20909,6 +20971,7 @@ async def _send_update_notification(self) -> bool: claimed_path.unlink(missing_ok=True) output_path.unlink(missing_ok=True) exit_code_path.unlink(missing_ok=True) + owner_result_path.unlink(missing_ok=True) return True diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index e719eb2ac88a..d87ca2b744c0 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -5427,6 +5427,12 @@ async def _handle_update_command(self, event: MessageEvent) -> str: pending_path = _hermes_home / ".update_pending.json" output_path = _hermes_home / ".update_output.txt" exit_code_path = _hermes_home / ".update_exit_code" + owner_restart_pending_path = ( + _hermes_home / ".update_owner_restart_pending.json" + ) + owner_restart_result_path = ( + _hermes_home / ".update_owner_restart_result.json" + ) session_key = self._session_key_for_source(event.source) pending = { "platform": event.source.platform.value, @@ -5444,6 +5450,8 @@ async def _handle_update_command(self, event: MessageEvent) -> str: _tmp_pending.write_text(json.dumps(pending), encoding="utf-8") _tmp_pending.replace(pending_path) exit_code_path.unlink(missing_ok=True) + owner_restart_pending_path.unlink(missing_ok=True) + owner_restart_result_path.unlink(missing_ok=True) # Spawn `hermes update --gateway` detached so it survives gateway restart. # --gateway enables file-based IPC for interactive prompts (stash @@ -5510,7 +5518,13 @@ async def _handle_update_command(self, event: MessageEvent) -> str: # in zsh, and this command string is copied/reused in macOS/zsh # operator wrappers. Keep the template zsh-safe even though this # specific subprocess currently runs under bash. - f"rc=$?; printf '%s' \"$rc\" > {shlex.quote(str(exit_code_path))}" + # + # A same-cgroup updater delegates the terminal restart and + # result marker to an external verifier. Never publish the + # updater shell's return code while that proof is pending. + f"rc=$?; if [ ! -e {shlex.quote(str(owner_restart_pending_path))} ] " + f"&& [ ! -e {shlex.quote(str(owner_restart_result_path))} ]; " + f"then printf '%s' \"$rc\" > {shlex.quote(str(exit_code_path))}; fi" ) setsid_bin = shutil.which("setsid") if setsid_bin: diff --git a/hermes_cli/dashboard_procs.py b/hermes_cli/dashboard_procs.py index 7379ac972e2e..b7d409875b8b 100644 --- a/hermes_cli/dashboard_procs.py +++ b/hermes_cli/dashboard_procs.py @@ -209,13 +209,34 @@ def _kill_stale_dashboard_processes( # stay a stop, not a restart. pid_cgroup: dict[int, str | None] = {} pid_service: dict[int, str | None] = {} + pid_service_snapshot: dict[int, dict] = {} pid_cmdline: dict[int, list[str]] = {} if restart_managed and sys.platform != "win32": for pid in pids: cg_path = _m()._get_pid_cgroup_path(pid) pid_cgroup[pid] = cg_path pid_service[pid] = _m()._get_systemd_service_for_pid(pid) - if not pid_service[pid]: + if pid_service[pid]: + scope_name = _m()._extract_scope_from_cgroup(cg_path) + if scope_name in {"user", "system"}: + scope = ("--user",) if scope_name == "user" else () + state = _m()._read_managed_dashboard_service_state( + scope, pid_service[pid] + ) + # The PID-to-cgroup proof is stronger than a transient + # is-active read failure. Preserve the killed PID as the + # old generation so post-restart verification cannot pass + # on an unchanged service process. + state["active"] = True + if int(state.get("main_pid") or 0) <= 0: + state["main_pid"] = pid + state.update( + scope=scope, + unit=pid_service[pid], + runtime=None, + ) + pid_service_snapshot[pid] = state + else: # Manually-started process: preserve its exact argv so we # can respawn it after the update (#40449, #68934). cmdline = _m()._dashboard_cmdline_for_pid(pid) @@ -295,6 +316,8 @@ def _kill_stale_dashboard_processes( # - manually-started PIDs: respawn the argv captured before the kill # (#40449) — detached, headless, logged to logs/dashboard-restart.log. restarted_services: list[str] = [] + restarted_service_snapshots: list[dict] = [] + unverified_services: list[str] = [] unrecovered: list[int] = [] if killed and restart_managed: failed_restarts: list[tuple[str, str]] = [] @@ -308,6 +331,11 @@ def _kill_stale_dashboard_processes( seen_services.add(svc_name) if _m()._try_restart_systemd_service(svc_name, pid_cgroup.get(pid)): restarted_services.append(svc_name) + snapshot = pid_service_snapshot.get(pid) + if snapshot is not None: + restarted_service_snapshots.append(snapshot) + else: + unverified_services.append(svc_name) else: failed_restarts.append((svc_name, "systemctl restart returned non-zero")) unrecovered.append(pid) @@ -339,6 +367,8 @@ def _kill_stale_dashboard_processes( "killed": list(killed), "failed": list(failed), "unrecovered": list(unrecovered), + "restarted_service_snapshots": list(restarted_service_snapshots), + "unverified_services": list(unverified_services), } def _detect_concurrent_hermes_instances( diff --git a/hermes_cli/main.py b/hermes_cli/main.py index cd9966cf8cff..44f4d2a7902c 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -7286,6 +7286,168 @@ def _dashboard_probe_host(host: str | None) -> str: _DASHBOARD_SYSTEMD_UNIT = "hermes-dashboard.service" +def _read_managed_dashboard_service_state( + scope: tuple[str, ...], + unit: str = _DASHBOARD_SYSTEMD_UNIT, +) -> dict: + """Read the managed dashboard's active/PID/start state from one scope.""" + active = subprocess.run( + ["systemctl", *scope, "is-active", unit], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=5, + ) + shown = subprocess.run( + [ + "systemctl", + *scope, + "show", + unit, + "--property=MainPID", + "--property=ActiveEnterTimestampMonotonic", + ], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=5, + ) + props: dict[str, str] = {} + for line in (shown.stdout or "").splitlines(): + key, sep, value = line.partition("=") + if sep: + props[key.strip()] = value.strip() + try: + main_pid = int(props.get("MainPID", "0") or 0) + except ValueError: + main_pid = 0 + try: + started = int(props.get("ActiveEnterTimestampMonotonic", "0") or 0) + except ValueError: + started = 0 + return { + "active": active.returncode == 0 and (active.stdout or "").strip() == "active", + "main_pid": main_pid, + "started": started, + } + + +def _snapshot_managed_dashboard_service( + unit: str = _DASHBOARD_SYSTEMD_UNIT, +) -> dict | None: + """Capture the managed dashboard state before updater cleanup. + + Returns ``None`` when the canonical unit is not installed or is inactive. + Runtime host and port are captured from the old PID when possible so + post-restart verification can include the public local ``/api/health`` + endpoint without starting a dashboard the operator had left stopped. + """ + if sys.platform == "win32": + return None + for scope in (("--user",), ()): + try: + listed = subprocess.run( + [ + "systemctl", + *scope, + "list-unit-files", + unit, + "--no-legend", + "--no-pager", + ], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=10, + ) + except (FileNotFoundError, subprocess.TimeoutExpired, OSError): + continue + rows = (listed.stdout or "").splitlines() + if listed.returncode != 0 or not any( + row.split()[0:1] == [unit] for row in rows if row.split() + ): + continue + try: + state = _read_managed_dashboard_service_state(scope, unit) + except (FileNotFoundError, subprocess.TimeoutExpired, OSError): + state = {"active": False, "main_pid": 0, "started": 0} + if not state["active"]: + # An installed/enabled unit can be intentionally stopped. The + # updater must not start a no-dashboard deployment merely because + # the unit file exists; keep looking in case the other scope owns + # the active canonical service. + continue + runtime = None + if state["main_pid"] > 0: + argv = _dashboard_cmdline_for_pid(state["main_pid"]) + if argv: + runtime = _parse_dashboard_runtime(shlex.join(argv)) + return {"scope": scope, "unit": unit, "runtime": runtime, **state} + return None + + +def _dashboard_healthcheck(runtime: tuple[str, str, int]) -> bool | None: + """Probe local dashboard health; return ``None`` when not safely supported.""" + import urllib.error + import urllib.request + + _mode, host, port = runtime + probe_host = _dashboard_probe_host(host) + if probe_host not in {"127.0.0.1", "localhost", "::1"}: + return None + url_host = f"[{probe_host}]" if ":" in probe_host else probe_host + try: + with urllib.request.urlopen( + f"http://{url_host}:{port}/api/health", timeout=2 + ) as response: + return 200 <= getattr(response, "status", 200) < 300 + except (OSError, TimeoutError, urllib.error.URLError): + return False + + +def _verify_managed_dashboard_restart( + before: dict, + *, + timeout: float = 20.0, +) -> bool: + """Boundedly verify active state, process generation, and local health.""" + deadline = _time.monotonic() + max(timeout, 0.0) + while True: + try: + after = _read_managed_dashboard_service_state( + tuple(before["scope"]), before["unit"] + ) + except (FileNotFoundError, subprocess.TimeoutExpired, OSError, KeyError): + after = {"active": False, "main_pid": 0, "started": 0} + + old_pid = int(before.get("main_pid") or 0) + old_started = int(before.get("started") or 0) + transitioned = bool( + after["active"] + and after["main_pid"] > 0 + and ( + old_pid <= 0 + or after["main_pid"] != old_pid + or ( + after["started"] > 0 + and after["started"] > old_started + ) + ) + ) + if transitioned: + health = None + if before.get("runtime") is not None: + health = _dashboard_healthcheck(before["runtime"]) + if health is not False: + return True + if _time.monotonic() >= deadline: + return False + _time.sleep(0.5) + + def _restart_managed_dashboard_service( reason: str, unit: str = _DASHBOARD_SYSTEMD_UNIT, diff --git a/hermes_cli/update_cmd.py b/hermes_cli/update_cmd.py index c9069b1b1a22..8f51423c2d8a 100644 --- a/hermes_cli/update_cmd.py +++ b/hermes_cli/update_cmd.py @@ -27,6 +27,7 @@ import json import logging import os +import secrets import shlex import shutil import subprocess @@ -567,25 +568,58 @@ def _format_time_ago(iso_ts: str) -> str: except Exception: return "recently" -def _finish_dashboard_update_cleanup(node_failures: list[str]) -> None: - """Refresh managed dashboards or stop stale manual ones after an update.""" +def _finish_dashboard_update_cleanup(node_failures: list[str]) -> bool: + """Refresh dashboards after an update and verify managed restarts. + + Returns ``True`` only when cleanup was intentionally skipped after a Node + refresh failure, no stale process needed recovery, or every required + restart was verified. Callers use the result for the durable update exit + marker instead of reporting success while an old backend is still live. + """ if node_failures: print() print(" ℹ Leaving running dashboard process(es) untouched because the") print(" Node.js dependency refresh did not complete.") - return + return True + managed_before = _m()._snapshot_managed_dashboard_service() stop_result = _m()._kill_stale_dashboard_processes(restart_managed=True) - if not stop_result.get("unrecovered"): - return - print() - print( - "⚠ A web dashboard/serve process was stopped during update and could " - "not be auto-restarted." - ) - print(" Re-launch it when you want the web UI back:") - print(" hermes dashboard --port ") + restart_snapshots: list[dict] = [] + if managed_before is not None: + restart_snapshots.append(managed_before) + restart_snapshots.extend(stop_result.get("restarted_service_snapshots") or []) + + unverified_services = list(stop_result.get("unverified_services") or []) + failed_verification: list[str] = [] + for snapshot in restart_snapshots: + if not _m()._verify_managed_dashboard_restart(snapshot): + failed_verification.append(str(snapshot.get("unit") or "dashboard service")) + failed_verification.extend(unverified_services) + if failed_verification: + print() + print( + "⚠ Managed dashboard restart could not be verified; a service may " + "still be running pre-update code." + ) + for unit in sorted(set(failed_verification)): + print(f" Check it with: systemctl status {unit}") + return False + + failed = bool(stop_result.get("failed")) + unrecovered = bool(stop_result.get("unrecovered")) + if not failed and not unrecovered: + return True + + if unrecovered: + print() + print( + "⚠ A web dashboard/serve process was stopped during update and could " + "not be auto-restarted." + ) + print(" Re-launch it when you want the web UI back:") + print(" hermes dashboard --port ") + return False def _atomic_replace_dir(src: str, dst: str) -> None: """Replace directory *dst* with *src* without leaving *dst* half-deleted. @@ -3253,13 +3287,17 @@ def _for_each_systemd_gateway_unit( *, process_unit, on_unit_timeout, -) -> None: + defer_unit: str | None = None, +) -> list[str]: """Process each ``hermes-gateway*.service`` from ``systemctl list-units``. ``subprocess.TimeoutExpired`` raised by ``process_unit`` is isolated to that unit via ``on_unit_timeout`` so one wedged systemctl call cannot - abort the rest of the fleet (#68523). + abort the rest of the fleet (#68523). ``defer_unit`` names the gateway + whose cgroup owns the updater; it is returned without being restarted so + the caller can make it the terminal action after all other finalization. """ + deferred: list[str] = [] for line in (list_units_stdout or "").strip().splitlines(): parts = line.split() if not parts: @@ -3272,10 +3310,339 @@ def _for_each_systemd_gateway_unit( if not unit.startswith("hermes-gateway"): continue svc_name = unit.removesuffix(".service") + if defer_unit is not None and svc_name == defer_unit: + deferred.append(svc_name) + continue try: process_unit(svc_name) except subprocess.TimeoutExpired as exc: on_unit_timeout(svc_name, exc) + return deferred + + +def _detect_updater_systemd_gateway_owner( +) -> tuple[tuple[str, list[str], str] | None, bool]: + """Return the systemd gateway unit whose cgroup contains this updater. + + The boolean reports a malformed gateway-looking cgroup that could not be + resolved safely. A missing/non-systemd cgroup is normal on macOS, + Windows, containers, and manually launched gateways, so it is not treated + as a discovery failure. + """ + cgroup_path = _m()._get_pid_cgroup_path(os.getpid()) + if not cgroup_path: + return None, False + cgroup_parts = [part for part in cgroup_path.split("/") if part] + gateway_units = [ + part + for part in cgroup_parts + if part.startswith("hermes-gateway") and part.endswith(".service") + ] + gateway_cgroup = bool(gateway_units) or ( + "hermes-gateway" in cgroup_path and ".service" in cgroup_path + ) + svc_name = _m()._get_systemd_service_for_pid(os.getpid()) + if not svc_name and gateway_units: + # ``_get_systemd_service_for_pid`` historically recognized only a + # cgroup path ending at the service. Delegated workers may live in a + # nested child scope below that service, so recover the exact owning + # unit component without weakening the hermes-gateway name gate. + svc_name = gateway_units[-1] + if not svc_name: + return None, gateway_cgroup + if not svc_name.startswith("hermes-gateway"): + return None, False + scope = _m()._extract_scope_from_cgroup(cgroup_path) + if scope not in {"user", "system"}: + return None, True + scope_cmd = ["systemctl", "--user"] if scope == "user" else ["systemctl"] + return (scope, scope_cmd, svc_name.removesuffix(".service")), False + + +def _write_gateway_update_exit_code(required: bool, code: int) -> bool: + """Atomically persist a terminal updater result when a consumer needs it.""" + if not required: + return True + try: + from utils import atomic_write_text + + atomic_write_text( + get_hermes_home() / ".update_exit_code", + str(int(code)), + ) + except OSError: + return False + return True + + +def _owner_restart_verification_timeout( + owner: tuple[str, list[str], str], +) -> float: + """Bound owner drain, restart backoff, and new-gateway startup verification.""" + scope, scope_cmd, svc_name = owner + try: + from hermes_cli.gateway import _get_restart_exit_wait_budget + + exit_wait_budget = max(0.0, float(_get_restart_exit_wait_budget())) + except Exception: + exit_wait_budget = 105.0 + prompt_free_scope_cmd = list(scope_cmd) + if "--no-ask-password" not in prompt_free_scope_cmd: + prompt_free_scope_cmd.append("--no-ask-password") + restart_probe = prompt_free_scope_cmd + [ + "show", + svc_name, + "--property=RestartUSec", + "--value", + ] + prompt_free_scope_cmd = _resolve_noninteractive_systemd_command( + scope, + prompt_free_scope_cmd, + targeted_probe=restart_probe, + ) + if prompt_free_scope_cmd is None: + raise PermissionError("no prompt-free systemctl capability for owner verifier") + restart_timeout = _service_restart_sec( + prompt_free_scope_cmd, svc_name, default=0.0 + ) + # SIGUSR1 may first wait for the active turn and only then enter stop/drain. + # Cover the same full wait budget used by gateway restart callers, plus + # service backoff and two minutes for replacement startup/readiness. Keep + # the verifier within the gateway update watcher's bounded 30-minute window. + return min(1800.0, max(180.0, exit_wait_budget + restart_timeout + 120.0)) + + +def _clear_owner_restart_request_files() -> None: + from hermes_cli.update_owner_restart import ( + OWNER_RESTART_ACK_FILE, + OWNER_RESTART_PENDING_FILE, + ) + + home = get_hermes_home() + for name in (OWNER_RESTART_PENDING_FILE, OWNER_RESTART_ACK_FILE): + try: + (home / name).unlink(missing_ok=True) + except OSError: + pass + + +def _resolve_noninteractive_systemd_command( + scope: str, + command: list[str], + *, + targeted_probe: list[str] | None = None, +) -> list[str] | None: + """Resolve a prompt-free user/system systemd command prefix. + + User-scope commands run directly. System-scope commands run directly as + root; otherwise they use ``sudo -n`` only after the same blanket/targeted + capability probes used by gateway manage-unit operations. + """ + if scope == "user": + return command + if scope != "system" or not hasattr(os, "geteuid"): + return None + if os.geteuid() == 0: + return command + + sudo_command = ["sudo", "-n"] + command + try: + probe = subprocess.run( + ["sudo", "-n", "true"], + capture_output=True, + timeout=5, + ) + if probe.returncode == 0: + return sudo_command + if targeted_probe is None: + return None + probe = subprocess.run( + ["sudo", "-n"] + targeted_probe, + capture_output=True, + timeout=5, + ) + except (FileNotFoundError, OSError, subprocess.TimeoutExpired): + return None + return sudo_command if probe.returncode == 0 else None + + +def _launch_updater_owner_restart_verifier( + owner: tuple[str, list[str], str], + *, + final_exit_code: int, + persist_result: bool, +) -> bool: + """Launch the terminal owner restart in a separate transient systemd cgroup. + + The transient verifier, not this updater-owned cgroup, sends SIGUSR1. It + then boundedly proves ActiveState, changed MainPID/start generation, and a + readiness acknowledgement from the new gateway before atomically exposing + the terminal update result. If the transient unit cannot be launched, the + owner remains untouched and the caller records an immediate failure. + """ + if not persist_result: + return False + systemd_run = shutil.which("systemd-run") + if not systemd_run: + return False + + scope, _scope_cmd, svc_name = owner + command = [systemd_run] + if scope == "user": + command.extend(["--user", "--no-ask-password"]) + else: + command.append("--no-ask-password") + command = _resolve_noninteractive_systemd_command( + scope, + command, + targeted_probe=command + ["--version"], + ) + if command is None: + return False + + from hermes_cli import update_owner_restart + + try: + old_state = update_owner_restart.read_systemd_service_state(scope, svc_name) + timeout_seconds = _owner_restart_verification_timeout(owner) + nonce = secrets.token_hex(16) + home = get_hermes_home() + update_owner_restart.prepare_owner_restart_request( + home, + scope=scope, + service=svc_name, + old_state=old_state, + final_exit_code=int(final_exit_code), + timeout_seconds=timeout_seconds, + nonce=nonce, + ) + except (FileNotFoundError, OSError, TypeError, ValueError, subprocess.TimeoutExpired): + return False + + unit_name = ( + f"hermes-update-owner-verify-{os.getpid()}-{nonce[:8]}".replace(".", "-") + ) + project_root = Path(__file__).resolve().parent.parent + command.extend( + [ + "--collect", + f"--unit={unit_name}", + "--property=Type=exec", + f"--property=RuntimeMaxSec={int(timeout_seconds + 30)}", + f"--property=WorkingDirectory={project_root}", + f"--setenv=HERMES_HOME={home}", + "--setenv=PYTHONUNBUFFERED=1", + ] + ) + if scope == "system": + command.extend( + [ + f"--property=User={os.geteuid()}", + f"--property=Group={os.getegid()}", + ] + ) + command.extend( + [ + sys.executable, + "-m", + "hermes_cli.update_owner_restart", + "--home", + str(home), + f"--nonce={nonce}", + ] + ) + + try: + launched = subprocess.run( + command, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=10, + ) + except (FileNotFoundError, OSError, subprocess.TimeoutExpired): + return False + if launched.returncode != 0: + _clear_owner_restart_request_files() + return False + return True + + +def _finish_update_service_finalization( + node_failures: list[str], + *, + gateway_mode: bool, + gateway_fleet_restart_incomplete: bool, + updater_owner_discovery_failed: bool, + deferred_owner: tuple[str, list[str], str] | None, +) -> bool: + """Finalize dashboard/result state, then restart the owner exactly last.""" + dashboard_ok = _finish_dashboard_update_cleanup(node_failures) + final_ok = ( + dashboard_ok + and not node_failures + and not gateway_fleet_restart_incomplete + and not updater_owner_discovery_failed + ) + result_required = gateway_mode or deferred_owner is not None + marker_ok = True + verifier_launched = False + if deferred_owner is not None: + print() + print( + f" → Handing final updater-owner restart for {deferred_owner[2]} " + "to an external verifier..." + ) + verifier_launched = _launch_updater_owner_restart_verifier( + deferred_owner, + final_exit_code=0 if final_ok else 1, + persist_result=result_required, + ) + if verifier_launched: + print( + " → Owner restart verification is pending outside the service " + "cgroup; completion will be published only after readiness proof." + ) + else: + final_ok = False + marker_ok = _write_gateway_update_exit_code(result_required, 1) + print( + f" ✗ Could not launch the external restart verifier for " + f"{deferred_owner[2]}." + ) + else: + marker_ok = _write_gateway_update_exit_code( + result_required, 0 if final_ok else 1 + ) + + if not marker_ok: + final_ok = False + print() + print( + " ✗ Could not persist the durable update result; leaving the " + "updater-owning service running." + ) + + if deferred_owner is not None and verifier_launched: + # This process belongs to the owner cgroup that the verifier will tear + # down. Only the external verifier can truthfully publish completion + # after observing the replacement process and its readiness ack. + return False + + print() + if node_failures: + print( + "⚠ Update partially complete — Node.js dependencies for " + f"{', '.join(node_failures)} did not refresh." + ) + print(" Code and Python deps are updated, but the dashboard/TUI may") + print(" be in a mixed state until the Node deps are rebuilt.") + elif final_ok: + print("✓ Update complete!") + else: + print("⚠ Update finalization incomplete — see the warnings above.") + return final_ok def _warn_incomplete_gateway_fleet_restart(failed_units: list) -> None: """Print an explicit incomplete-update warning for unrestarted units.""" @@ -4582,16 +4949,9 @@ def _print_items(items, label, key, fallback_key=None): # Never let the cron safety net break an otherwise-good update. logger.debug("Cron jobs auto-restore check failed: %s", exc) - print() - if node_failures: - print( - "⚠ Update partially complete — Node.js dependencies for " - f"{', '.join(node_failures)} did not refresh." - ) - print(" Code and Python deps are updated, but the dashboard/TUI may") - print(" be in a mixed state until the Node deps are rebuilt.") - else: - print("✓ Update complete!") + # The final success/partial status is emitted only after gateway and + # dashboard finalization. Printing it here used to produce false + # success when a later restart failed or killed this updater. # Search-index optimization notice (v23). Existing installs keep their # working search index untouched on update; the compact v23 layout — @@ -4681,30 +5041,17 @@ def _print_items(items, label, key, fallback_key=None): except Exception as e: logger.debug("cua-driver refresh failed: %s", e) - # Write exit code *before* the gateway restart attempt. - # When running as ``hermes update --gateway`` (spawned by the gateway's - # /update command), this process lives inside the gateway's systemd - # cgroup. A graceful SIGUSR1 restart keeps the drain loop alive long - # enough for the exit-code marker to be written below, but the - # fallback ``systemctl restart`` path (see below) kills everything in - # the cgroup (KillMode=mixed → SIGKILL to remaining processes), - # including us and the wrapping bash shell. The shell never reaches - # its ``printf $status > .update_exit_code`` epilogue, so the - # exit-code marker file would never be created. The new gateway's - # update watcher would then poll for 30 minutes and send a spurious - # timeout message. - # - # Writing the marker here — after git pull + pip install succeed but - # before we attempt the restart — ensures the new gateway sees it - # regardless of how we die. - if gateway_mode: - _exit_code_path = get_hermes_home() / ".update_exit_code" - try: - _exit_code_path.write_text("0", encoding="utf-8") - except OSError: - pass + # Do not publish an exit marker yet. The gateway watcher treats the + # marker's mere presence as terminal, so even an optimistic ``0`` can + # be consumed before gateway/dashboard finalization discovers a later + # failure. Ownership-aware finalization below writes the sole result + # immediately before the terminal owner restart. gateway_fleet_restart_incomplete = False + updater_owner, updater_owner_discovery_failed = ( + _detect_updater_systemd_gateway_owner() + ) + deferred_owner: tuple[str, list[str], str] | None = None # Auto-restart ALL gateways after update. # The code update (git pull) is shared across all profiles, so every @@ -4833,33 +5180,11 @@ def _resolve_manage_cmd(scope_: str, scope_cmd_: list, svc_name_: str): if scope_ in _manage_cmd_cache: return _manage_cmd_cache[scope_] cmd = scope_cmd_ + ["--no-ask-password"] - if ( - scope_ == "system" - and hasattr(os, "geteuid") - and os.geteuid() != 0 # windows-footgun: ok — systemd path, Linux-only - ): - sudo_cmd = ["sudo", "-n"] + scope_cmd_ + ["--no-ask-password"] - sudo_ok = False - try: - _probe = subprocess.run( - ["sudo", "-n", "true"], - capture_output=True, - timeout=5, - ) - sudo_ok = _probe.returncode == 0 - if not sudo_ok: - # Blanket sudo refused — a targeted sudoers entry - # (NOPASSWD for systemctl ... hermes-gateway*) - # may still allow the exact commands we need. - _probe = subprocess.run( - sudo_cmd + ["reset-failed", svc_name_], - capture_output=True, - timeout=5, - ) - sudo_ok = _probe.returncode == 0 - except (FileNotFoundError, subprocess.TimeoutExpired): - sudo_ok = False - cmd = sudo_cmd if sudo_ok else None + cmd = _resolve_noninteractive_systemd_command( + scope_, + cmd, + targeted_probe=cmd + ["reset-failed", svc_name_], + ) _manage_cmd_cache[scope_] = cmd return cmd @@ -4885,8 +5210,18 @@ def _resolve_manage_cmd(scope_: str, scope_cmd_: list, svc_name_: str): externally_supervised_profiles = [] # --- Systemd services (Linux) --- - # Discover all hermes-gateway* units (default + profiles) - if supports_systemd_services(): + # Discover all hermes-gateway* units (default + profiles). A + # gateway-looking updater cgroup with no safely resolved owner is + # fail-closed: restarting any candidate could kill this updater + # before dashboard cleanup and the durable failure marker. + if supports_systemd_services() and updater_owner_discovery_failed: + gateway_fleet_restart_incomplete = True + print() + print( + " ⚠ Could not safely resolve the updater's systemd gateway " + "owner; leaving systemd gateway units running." + ) + if supports_systemd_services() and not updater_owner_discovery_failed: try: _ensure_user_systemd_env() except Exception: @@ -5178,10 +5513,28 @@ def _on_unit_timeout(svc_name: str, exc: subprocess.TimeoutExpired) -> None: f"continuing with remaining gateways" ) - _for_each_systemd_gateway_unit( + _defer_unit = None + if updater_owner is not None and updater_owner[0] == scope: + _defer_unit = updater_owner[2] + _deferred_units = _for_each_systemd_gateway_unit( result.stdout, process_unit=_restart_one_systemd_gateway_unit, on_unit_timeout=_on_unit_timeout, + defer_unit=_defer_unit, + ) + if _deferred_units: + deferred_owner = ( + scope, + list(scope_cmd), + _deferred_units[0], + ) + + if updater_owner is not None and deferred_owner is None: + updater_owner_discovery_failed = True + print() + print( + " ⚠ The updater's systemd gateway owner was detected " + "but not present in the restart inventory." ) # --- Launchd services (macOS) --- @@ -5310,12 +5663,6 @@ def _on_unit_timeout(svc_name: str, exc: subprocess.TimeoutExpired) -> None: if failed_or_stale_units: gateway_fleet_restart_incomplete = True - if gateway_mode: - _exit_code_path = get_hermes_home() / ".update_exit_code" - try: - _exit_code_path.write_text("1", encoding="utf-8") - except OSError: - pass _warn_incomplete_gateway_fleet_restart(failed_or_stale_units) if not restarted_services and not killed_pids: @@ -5365,7 +5712,13 @@ def _on_unit_timeout(svc_name: str, exc: subprocess.TimeoutExpired) -> None: logger.debug("Post-restart survivor sweep failed: %s", _sweep_exc) except Exception as e: + gateway_fleet_restart_incomplete = True logger.debug("Gateway restart during update failed: %s", e) + print() + print(f" ⚠ Gateway restart finalization failed: {e}") + + if updater_owner is not None and deferred_owner is None: + updater_owner_discovery_failed = True _m()._resume_windows_gateways_after_update(_windows_gateway_resume) @@ -5398,21 +5751,21 @@ def _on_unit_timeout(svc_name: str, exc: subprocess.TimeoutExpired) -> None: except Exception as e: logger.debug("Legacy unit check during update failed: %s", e) - # Restart a managed dashboard through systemd, or stop stale manual - # dashboard processes. Raw-killing a systemd-owned dashboard PID makes - # systemd treat it as a clean stop, leaving the Cloudflare origin dead. - # Preserve the safety rule above: a failed Node refresh leaves the - # currently running dashboard untouched. - _finish_dashboard_update_cleanup(node_failures) - print() print("Tip: You can now select a provider and model:") print(" hermes model # Select provider and model") - if gateway_fleet_restart_incomplete: - # Code update itself succeeded, but at least one gateway still - # runs pre-update modules — surface that as a failed update so - # automation / operators do not treat the fleet as healthy. + # Dashboard cleanup and the durable result marker must precede the + # updater-owning service restart. Signalling that service is the + # terminal side effect because systemd may reap this updater's cgroup. + finalization_ok = _finish_update_service_finalization( + node_failures, + gateway_mode=gateway_mode, + gateway_fleet_restart_incomplete=gateway_fleet_restart_incomplete, + updater_owner_discovery_failed=updater_owner_discovery_failed, + deferred_owner=deferred_owner, + ) + if not finalization_ok: sys.exit(1) except subprocess.CalledProcessError as e: diff --git a/hermes_cli/update_owner_restart.py b/hermes_cli/update_owner_restart.py new file mode 100644 index 000000000000..4e67f93d6df7 --- /dev/null +++ b/hermes_cli/update_owner_restart.py @@ -0,0 +1,662 @@ +"""Durable, out-of-cgroup verification for updater-owned gateway restarts. + +The updater can run inside the gateway service it must restart. It therefore +cannot wait for that service's cgroup to disappear and return without risking +its own teardown. This module is launched as a transient systemd service in a +separate cgroup. It owns the final SIGUSR1, verifies a changed systemd process +and start generation plus an in-process readiness acknowledgement, then writes +the terminal update result atomically. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import signal +import subprocess +import time +from pathlib import Path +from typing import Any + +from utils import atomic_json_write, atomic_write_text + +OWNER_RESTART_PENDING_FILE = ".update_owner_restart_pending.json" +OWNER_RESTART_ACK_FILE = ".update_owner_restart_ack.json" +OWNER_RESTART_RESULT_FILE = ".update_owner_restart_result.json" +_OWNER_RESTART_LOCK_FILE = ".update_owner_restart_verifier.lock" +_UPDATE_EXIT_CODE_FILE = ".update_exit_code" +_UPDATE_OUTPUT_FILE = ".update_output.txt" +_REQUEST_VERSION = 2 +_RESULT_VERSION = 1 +_NONCE_RE = re.compile(r"[0-9a-f]{32}") +_RESTART_NO_EXIT_REASON = "owner exited and Restart=no disables automatic restart" + + +def _marker(home: Path, name: str) -> Path: + return Path(home) / name + + +def _scope_command(scope: str) -> list[str]: + if scope == "user": + return ["systemctl", "--user", "--no-ask-password"] + if scope == "system": + return ["systemctl", "--no-ask-password"] + raise ValueError(f"unsupported systemd scope: {scope!r}") + + +_SYSTEMD_STATE_ARGS = [ + "--property=ActiveState", + "--property=SubState", + "--property=MainPID", + "--property=ExecMainStartTimestampMonotonic", + "--property=ActiveEnterTimestampMonotonic", + "--property=Restart", +] + + +def _run_systemd_state_query(scope: str, service: str) -> subprocess.CompletedProcess[str]: + command = _scope_command(scope) + ["show", service, *_SYSTEMD_STATE_ARGS] + run_kwargs = { + "capture_output": True, + "text": True, + "encoding": "utf-8", + "errors": "replace", + "timeout": 5, + } + if scope == "user" or (hasattr(os, "geteuid") and os.geteuid() == 0): + return subprocess.run(command, **run_kwargs) + if not hasattr(os, "geteuid"): + raise OSError("system systemd scope requires a prompt-free privilege path") + + # Match the updater's existing noninteractive policy: prefer a blanket + # passwordless capability, but still try the exact read-only command for a + # targeted sudoers rule. ``-n`` and ``--no-ask-password`` prohibit both + # sudo and polkit prompts. + subprocess.run( + ["sudo", "-n", "true"], + capture_output=True, + timeout=5, + ) + return subprocess.run(["sudo", "-n", *command], **run_kwargs) + + +def _coerce_int(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + +def _valid_service_name(service: Any) -> bool: + return ( + isinstance(service, str) + and service.startswith("hermes-gateway") + and "/" not in service + and "\\" not in service + and not service.endswith(".service") + ) + + +def _validate_nonce(nonce: Any) -> str: + if not isinstance(nonce, str) or _NONCE_RE.fullmatch(nonce) is None: + raise ValueError("invalid owner restart nonce") + return nonce + + +def _request_generation(request: dict[str, Any]) -> tuple[str, int]: + key = request.get("generation_key") + if key not in {"exec_start", "active_enter"}: + raise ValueError("invalid owner restart generation key") + value = _coerce_int(request.get("old_state", {}).get(key)) + if value <= 0: + raise ValueError("missing owner restart start generation") + return str(key), value + + +def _validate_request(payload: Any) -> dict[str, Any]: + if not isinstance(payload, dict) or payload.get("version") != _REQUEST_VERSION: + raise ValueError("unsupported owner restart request") + _validate_nonce(payload.get("nonce")) + if payload.get("scope") not in {"user", "system"}: + raise ValueError("invalid owner restart scope") + if not _valid_service_name(payload.get("service")): + raise ValueError("invalid owner restart service") + old_state = payload.get("old_state") + if not isinstance(old_state, dict): + raise ValueError("missing owner restart state") + if ( + old_state.get("active_state") != "active" + or old_state.get("sub_state") != "running" + or _coerce_int(old_state.get("main_pid")) <= 0 + ): + raise ValueError("owner service was not active and running") + _request_generation(payload) + if _coerce_int(payload.get("requested_at_ns")) <= 0: + raise ValueError("invalid owner restart request timestamp") + if _coerce_int(payload.get("deadline_ns")) <= _coerce_int( + payload.get("requested_at_ns") + ): + raise ValueError("invalid owner restart deadline") + if payload.get("final_exit_code") not in {0, 1}: + raise ValueError("invalid deferred update exit code") + return payload + + +def _read_json(path: Path) -> Any: + return json.loads(path.read_text(encoding="utf-8")) + + +def prepare_owner_restart_request( + home: Path, + *, + scope: str, + service: str, + old_state: dict[str, Any], + final_exit_code: int, + timeout_seconds: float, + nonce: str, +) -> dict[str, Any]: + """Atomically stage one owner restart request and clear stale IPC state.""" + home = Path(home) + _scope_command(scope) + if not _valid_service_name(service): + raise ValueError("invalid owner restart service") + nonce = _validate_nonce(nonce) + timeout_seconds = float(timeout_seconds) + if timeout_seconds <= 0 or timeout_seconds > 1800: + raise ValueError("owner restart timeout must be in (0, 1800]") + + normalized_state = { + "active_state": str(old_state.get("active_state") or ""), + "sub_state": str(old_state.get("sub_state") or ""), + "main_pid": _coerce_int(old_state.get("main_pid")), + "exec_start": _coerce_int(old_state.get("exec_start")), + "active_enter": _coerce_int(old_state.get("active_enter")), + "restart": str(old_state.get("restart") or ""), + } + generation_key = ( + "exec_start" if normalized_state["exec_start"] > 0 else "active_enter" + ) + requested_at_ns = time.time_ns() + request = { + "version": _REQUEST_VERSION, + "nonce": nonce, + "scope": scope, + "service": service, + "old_state": normalized_state, + "generation_key": generation_key, + "final_exit_code": int(final_exit_code), + "requested_at_ns": requested_at_ns, + "deadline_ns": requested_at_ns + int(timeout_seconds * 1_000_000_000), + "timeout_seconds": timeout_seconds, + } + _validate_request(request) + + for name in ( + OWNER_RESTART_ACK_FILE, + OWNER_RESTART_RESULT_FILE, + _OWNER_RESTART_LOCK_FILE, + ): + try: + _marker(home, name).unlink(missing_ok=True) + except OSError: + pass + atomic_json_write( + _marker(home, OWNER_RESTART_PENDING_FILE), request, indent=2, mode=0o600 + ) + return request + + +def read_systemd_service_state(scope: str, service: str) -> dict[str, Any]: + """Read one systemd unit's lifecycle identity in a single bounded query.""" + shown = _run_systemd_state_query(scope, service) + if shown.returncode != 0: + raise OSError((shown.stderr or "systemctl show failed").strip()) + props: dict[str, str] = {} + for line in (shown.stdout or "").splitlines(): + key, sep, value = line.partition("=") + if sep: + props[key.strip()] = value.strip() + return { + "active_state": props.get("ActiveState", ""), + "sub_state": props.get("SubState", ""), + "main_pid": _coerce_int(props.get("MainPID")), + "exec_start": _coerce_int(props.get("ExecMainStartTimestampMonotonic")), + "active_enter": _coerce_int( + props.get("ActiveEnterTimestampMonotonic") + ), + "restart": props.get("Restart", ""), + } + + +def current_systemd_gateway_service() -> str | None: + """Return the exact gateway service component containing this process.""" + try: + cgroup_text = Path("/proc/self/cgroup").read_text( + encoding="utf-8", errors="replace" + ) + except OSError: + return None + for line in cgroup_text.splitlines(): + cgroup_path = line.split(":", 2)[-1] + for component in reversed(cgroup_path.split("/")): + if component.startswith("hermes-gateway") and component.endswith( + ".service" + ): + return component.removesuffix(".service") + return None + + +def acknowledge_owner_restart_ready( + home: Path, + *, + current_service: str | None = None, + current_pid: int | None = None, + now_ns: int | None = None, +) -> bool: + """Acknowledge readiness only from the new, matching gateway owner. + + Call this after the gateway has connected/configured its adapters and marked + its runtime state running. The external verifier still independently checks + systemd ActiveState, MainPID, and start generation before trusting this ack. + """ + home = Path(home) + pending_path = _marker(home, OWNER_RESTART_PENDING_FILE) + if not pending_path.exists(): + return False + try: + request = _validate_request(_read_json(pending_path)) + current_service = current_service or current_systemd_gateway_service() + current_pid = int(current_pid if current_pid is not None else os.getpid()) + now_ns = int(now_ns if now_ns is not None else time.time_ns()) + if current_service != request["service"]: + return False + if current_pid <= 0 or current_pid == request["old_state"]["main_pid"]: + return False + if now_ns < request["requested_at_ns"] or now_ns > request["deadline_ns"]: + return False + atomic_json_write( + _marker(home, OWNER_RESTART_ACK_FILE), + { + "version": 1, + "nonce": request["nonce"], + "service": current_service, + "pid": current_pid, + "acknowledged_at_ns": now_ns, + }, + indent=2, + mode=0o600, + ) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + return False + return True + + +def _matching_ready_ack( + home: Path, request: dict[str, Any], state: dict[str, Any] +) -> bool: + try: + ack = _read_json(_marker(home, OWNER_RESTART_ACK_FILE)) + return bool( + isinstance(ack, dict) + and ack.get("version") == 1 + and ack.get("nonce") == request["nonce"] + and ack.get("service") == request["service"] + and _coerce_int(ack.get("pid")) == _coerce_int(state.get("main_pid")) + and request["requested_at_ns"] + <= _coerce_int(ack.get("acknowledged_at_ns")) + <= request["deadline_ns"] + ) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + return False + + +def _transition_verified( + home: Path, request: dict[str, Any], state: dict[str, Any] +) -> bool: + key, old_generation = _request_generation(request) + return bool( + state.get("active_state") == "active" + and state.get("sub_state") == "running" + and _coerce_int(state.get("main_pid")) > 0 + and _coerce_int(state.get("main_pid")) + != _coerce_int(request["old_state"].get("main_pid")) + and _coerce_int(state.get(key)) > 0 + and _coerce_int(state.get(key)) != old_generation + and _matching_ready_ack(home, request, state) + ) + + +def _append_update_output(home: Path, message: str) -> None: + path = _marker(home, _UPDATE_OUTPUT_FILE) + path.parent.mkdir(parents=True, exist_ok=True) + prefix = "" + try: + if path.exists() and path.stat().st_size > 0: + prefix = "\n" + except OSError: + prefix = "\n" + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + try: + os.write(fd, f"{prefix}{message.rstrip()}\n".encode("utf-8")) + os.fsync(fd) + finally: + os.close(fd) + + +def _result_message( + request: dict[str, Any], *, verified: bool, exit_code: int, reason: str +) -> str: + if not verified and reason == _RESTART_NO_EXIT_REASON: + command = "systemctl --user" if request["scope"] == "user" else "sudo systemctl" + return ( + "✗ Update finalization incomplete: updater-owning service " + f"{request['service']} exited and has Restart=no; Hermes did not auto-start it.\n" + " Restart it manually to load the updated code:\n" + f" {command} start {request['service']}\n" + " Then verify:\n" + f" {command} status {request['service']}" + ) + scope_flag = "--user " if request["scope"] == "user" else "" + if verified and exit_code == 0: + return ( + "✓ Update complete! Updater-owning service " + f"{request['service']} restarted and passed readiness verification." + ) + if verified: + return ( + "⚠ Update finalization incomplete before the owner restart; " + f"{request['service']} itself restarted successfully." + ) + return ( + "✗ Update finalization incomplete: updater-owning service " + f"{request['service']} restart verification failed ({reason}).\n" + f" Check: systemctl {scope_flag}status {request['service']}\n" + f" journalctl {scope_flag}-u {request['service']} --since '10 min ago'" + ) + + +def _validate_result(payload: Any) -> dict[str, Any]: + if not isinstance(payload, dict) or payload.get("version") != _RESULT_VERSION: + raise ValueError("unsupported owner restart result") + _validate_nonce(payload.get("nonce")) + if not _valid_service_name(payload.get("service")): + raise ValueError("invalid owner restart result service") + if not isinstance(payload.get("verified"), bool): + raise ValueError("invalid owner restart verification flag") + if payload.get("exit_code") not in {0, 1}: + raise ValueError("invalid owner restart result exit code") + if payload.get("exit_code") == 0 and payload.get("verified") is not True: + raise ValueError("unverified owner restart cannot succeed") + if not isinstance(payload.get("reason"), str) or not isinstance( + payload.get("message"), str + ): + raise ValueError("invalid owner restart result detail") + if _coerce_int(payload.get("completed_at_ns")) <= 0: + raise ValueError("invalid owner restart completion timestamp") + return payload + + +def read_owner_restart_result_exit_code(home: Path) -> int | None: + """Return the atomically committed owner result, if it is valid.""" + try: + result = _validate_result( + _read_json(_marker(Path(home), OWNER_RESTART_RESULT_FILE)) + ) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + return None + return int(result["exit_code"]) + + +def _persist_verifier_result( + home: Path, + request: dict[str, Any], + *, + verified: bool, + reason: str, + observed_state: dict[str, Any], +) -> int: + exit_code = int(request["final_exit_code"]) if verified else 1 + message = _result_message( + request, verified=verified, exit_code=exit_code, reason=reason + ) + result = { + "version": _RESULT_VERSION, + "nonce": request["nonce"], + "service": request["service"], + "verified": bool(verified), + "exit_code": exit_code, + "reason": reason, + "observed_state": observed_state, + "completed_at_ns": time.time_ns(), + "message": message, + } + + # Prepare human-readable evidence first, then atomically commit the JSON + # result as the owner path's authoritative terminal record. The legacy + # .update_exit_code marker is only a compatibility projection; gateway + # watchers also consume the JSON result directly if that projection fails. + try: + _append_update_output(home, message) + except OSError: + pass + try: + atomic_json_write( + _marker(home, OWNER_RESTART_RESULT_FILE), + result, + indent=2, + mode=0o600, + ) + except (OSError, TypeError, ValueError): + return 1 + + try: + atomic_write_text(_marker(home, _UPDATE_EXIT_CODE_FILE), str(exit_code)) + except OSError: + # The atomic JSON result above remains authoritative and recoverable by + # the restarted gateway even when this compatibility marker is unwritable. + pass + return exit_code + + +def _existing_result(home: Path, nonce: str) -> int | None: + try: + result = _validate_result( + _read_json(_marker(home, OWNER_RESTART_RESULT_FILE)) + ) + if result.get("nonce") == nonce: + return int(result["exit_code"]) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + pass + return None + + +def _acquire_verifier_lock(home: Path, nonce: str) -> int | None: + path = _marker(home, _OWNER_RESTART_LOCK_FILE) + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + return None + os.write(fd, nonce.encode("ascii")) + os.fsync(fd) + return fd + + +def verify_owner_restart( + home: Path, nonce: str, *, poll_interval: float = 0.25 +) -> int: + """Signal and boundedly verify one staged owner restart request.""" + home = Path(home) + try: + nonce = _validate_nonce(nonce) + except ValueError: + return 1 + existing = _existing_result(home, nonce) + if existing is not None: + return existing + + try: + request = _validate_request( + _read_json(_marker(home, OWNER_RESTART_PENDING_FILE)) + ) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + return 1 + if request["nonce"] != nonce: + return 1 + + lock_fd = _acquire_verifier_lock(home, nonce) + if lock_fd is None: + existing = _existing_result(home, nonce) + return existing if existing is not None else 1 + os.close(lock_fd) + + state: dict[str, Any] = {} + try: + try: + state = read_systemd_service_state(request["scope"], request["service"]) + except (FileNotFoundError, OSError, subprocess.TimeoutExpired): + return _persist_verifier_result( + home, + request, + verified=False, + reason="could not read initial systemd owner state", + observed_state=state, + ) + + old_pid = _coerce_int(request["old_state"].get("main_pid")) + restart_no = request["old_state"].get("restart") == "no" + initial_old_gone = bool( + _coerce_int(state.get("main_pid")) != old_pid + and state.get("active_state") not in {"activating", "deactivating"} + ) + if restart_no and initial_old_gone: + return _persist_verifier_result( + home, + request, + verified=False, + reason=_RESTART_NO_EXIT_REASON, + observed_state=state, + ) + if _transition_verified(home, request, state): + return _persist_verifier_result( + home, + request, + verified=True, + reason="owner transition and readiness acknowledged", + observed_state=state, + ) + if ( + state.get("active_state") != "active" + or state.get("sub_state") != "running" + or _coerce_int(state.get("main_pid")) != old_pid + ): + return _persist_verifier_result( + home, + request, + verified=False, + reason="owner state changed before verifier armed", + observed_state=state, + ) + + try: + os.kill(old_pid, signal.SIGUSR1) + except ProcessLookupError: + pass + except (PermissionError, OSError): + return _persist_verifier_result( + home, + request, + verified=False, + reason="could not signal old owner MainPID", + observed_state=state, + ) + + remaining = max( + 0.0, (request["deadline_ns"] - time.time_ns()) / 1_000_000_000 + ) + deadline = time.monotonic() + remaining + saw_transition_without_ack = False + saw_failed_state = False + while time.monotonic() < deadline: + try: + state = read_systemd_service_state( + request["scope"], request["service"] + ) + except (FileNotFoundError, OSError, subprocess.TimeoutExpired): + time.sleep(max(0.001, poll_interval)) + continue + + key, old_generation = _request_generation(request) + changed_identity = bool( + _coerce_int(state.get("main_pid")) > 0 + and _coerce_int(state.get("main_pid")) != old_pid + and _coerce_int(state.get(key)) > 0 + and _coerce_int(state.get(key)) != old_generation + ) + if ( + state.get("active_state") == "active" + and state.get("sub_state") == "running" + and changed_identity + ): + saw_transition_without_ack = True + if state.get("active_state") == "failed": + saw_failed_state = True + + old_gone = bool( + _coerce_int(state.get("main_pid")) != old_pid + and state.get("active_state") not in {"activating", "deactivating"} + ) + if restart_no and old_gone: + return _persist_verifier_result( + home, + request, + verified=False, + reason=_RESTART_NO_EXIT_REASON, + observed_state=state, + ) + + if _transition_verified(home, request, state): + return _persist_verifier_result( + home, + request, + verified=True, + reason="owner transition and readiness acknowledged", + observed_state=state, + ) + + time.sleep(max(0.001, poll_interval)) + + if saw_transition_without_ack: + reason = "new owner never wrote a matching readiness acknowledgement" + elif saw_failed_state: + reason = "owner entered failed state before restart verification" + else: + reason = "owner restart verification timed out without a transition" + return _persist_verifier_result( + home, + request, + verified=False, + reason=reason, + observed_state=state, + ) + finally: + try: + _marker(home, OWNER_RESTART_PENDING_FILE).unlink(missing_ok=True) + _marker(home, OWNER_RESTART_ACK_FILE).unlink(missing_ok=True) + _marker(home, _OWNER_RESTART_LOCK_FILE).unlink(missing_ok=True) + except OSError: + pass + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument("--home", required=True) + parser.add_argument("--nonce", required=True) + args = parser.parse_args(argv) + return verify_owner_restart(Path(args.home), args.nonce) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/gateway/test_update_command.py b/tests/gateway/test_update_command.py index a56dec11d80d..5e9a74b4b476 100644 --- a/tests/gateway/test_update_command.py +++ b/tests/gateway/test_update_command.py @@ -4,7 +4,9 @@ the _send_update_notification startup hook (sends results after restart). """ +import inspect import json +import os from pathlib import Path from unittest.mock import patch, MagicMock, AsyncMock @@ -169,8 +171,12 @@ def which_no_setsid(x): # Verify plain bash -c fallback (no nohup, no setsid) call_args = mock_popen.call_args[0][0] assert call_args[0] == "bash" - assert "nohup" not in call_args[2] - assert ".update_exit_code" in call_args[2] + script = call_args[2] + assert "nohup" not in script + assert ".update_exit_code" in script + assert ".update_owner_restart_pending.json" in script + assert ".update_owner_restart_result.json" in script + assert "if [ ! -e" in script # start_new_session=True should be in kwargs call_kwargs = mock_popen.call_args[1] assert call_kwargs.get("start_new_session") is True @@ -264,6 +270,30 @@ async def test_allows_homeassistant_via_registry_fallback(self, monkeypatch): class TestSendUpdateNotification: """Tests for GatewayRunner._send_update_notification.""" + def test_owner_readiness_ack_follows_running_state_and_precedes_notification(self): + from gateway.run import GatewayRunner + + source = inspect.getsource(GatewayRunner.start) + running = source.index('self._update_runtime_status("running")') + acknowledged = source.index("acknowledge_owner_restart_ready") + notification = source.index("self._send_update_notification()") + + assert running < acknowledged < notification + + def test_current_systemd_gateway_service_reads_nested_owner_cgroup(self): + from gateway import run as gateway_run + + cgroup = ( + "0::/user.slice/user-1000.slice/user@1000.service/app.slice/" + "hermes-gateway-coding_lead.service/worker.scope\n" + ) + with patch.object(gateway_run.sys, "platform", "linux"), patch.object( + gateway_run.Path, "read_text", return_value=cgroup + ): + assert ( + gateway_run._current_systemd_gateway_service() + == "hermes-gateway-coding_lead" + ) @pytest.mark.asyncio async def test_defers_notification_while_update_still_running(self, tmp_path): @@ -288,6 +318,164 @@ async def test_defers_notification_while_update_still_running(self, tmp_path): mock_adapter.send.assert_not_called() assert pending_path.exists() + @pytest.mark.asyncio + async def test_notifier_waits_for_external_verifier_terminal_result(self, tmp_path): + runner = _make_runner() + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + pending_path = hermes_home / ".update_pending.json" + pending_path.write_text(json.dumps({ + "platform": "telegram", "chat_id": "67890", "user_id": "12345", + })) + (hermes_home / ".update_output.txt").write_text("verification pending") + (hermes_home / ".update_owner_restart_ack.json").write_text(json.dumps({ + "version": 1, + "nonce": "a" * 32, + "service": "hermes-gateway-default", + "pid": os.getpid(), + })) + + mock_adapter = AsyncMock() + runner.adapters = {Platform.TELEGRAM: mock_adapter} + with patch("gateway.run._hermes_home", hermes_home): + assert await runner._send_update_notification() is False + mock_adapter.send.assert_not_awaited() + assert pending_path.exists() + + # Only the out-of-cgroup verifier may expose this terminal marker. + (hermes_home / ".update_exit_code").write_text("0") + assert await runner._send_update_notification() is True + + mock_adapter.send.assert_awaited_once() + assert not pending_path.exists() + + @pytest.mark.asyncio + async def test_atomic_owner_result_is_terminal_without_legacy_exit_marker( + self, tmp_path + ): + runner = _make_runner() + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + (hermes_home / ".update_pending.json").write_text( + json.dumps( + { + "platform": "telegram", + "chat_id": "67890", + "user_id": "12345", + } + ) + ) + (hermes_home / ".update_output.txt").write_text("owner verified") + result_path = hermes_home / ".update_owner_restart_result.json" + result_path.write_text( + json.dumps( + { + "version": 1, + "nonce": "a" * 32, + "service": "hermes-gateway-default", + "verified": True, + "exit_code": 0, + "reason": "owner transition and readiness acknowledged", + "observed_state": {"main_pid": 9876}, + "completed_at_ns": 1, + "message": "✓ Update complete!", + } + ) + ) + mock_adapter = AsyncMock() + runner.adapters = {Platform.TELEGRAM: mock_adapter} + + with patch("gateway.run._hermes_home", hermes_home): + assert await runner._send_update_notification() is True + + mock_adapter.send.assert_awaited_once() + assert not result_path.exists() + + @pytest.mark.asyncio + async def test_old_owner_does_not_publish_deferred_success(self, tmp_path): + runner = _make_runner() + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + pending_path = hermes_home / ".update_pending.json" + pending_path.write_text(json.dumps({ + "platform": "telegram", "chat_id": "67890", "user_id": "12345", + })) + owner_pending = hermes_home / ".update_owner_restart_pending.json" + owner_pending.write_text(json.dumps({ + "version": 1, + "exit_code": 0, + "owner_pid": os.getpid(), + "owner_service": "hermes-gateway-default", + })) + + mock_adapter = AsyncMock() + runner.adapters = {Platform.TELEGRAM: mock_adapter} + with patch("gateway.run._hermes_home", hermes_home), patch( + "gateway.run._current_systemd_gateway_service", + return_value="hermes-gateway-default", + create=True, + ): + assert await runner._send_update_notification() is False + + mock_adapter.send.assert_not_awaited() + assert owner_pending.exists() + assert pending_path.exists() + assert not (hermes_home / ".update_exit_code").exists() + + @pytest.mark.asyncio + async def test_other_gateway_cannot_promote_deferred_result(self, tmp_path): + runner = _make_runner() + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + pending_path = hermes_home / ".update_pending.json" + pending_path.write_text(json.dumps({ + "platform": "telegram", "chat_id": "67890", "user_id": "12345", + })) + owner_pending = hermes_home / ".update_owner_restart_pending.json" + owner_pending.write_text(json.dumps({ + "version": 1, + "exit_code": 0, + "owner_pid": os.getpid() + 1000, + "owner_service": "hermes-gateway-coding_lead", + })) + + mock_adapter = AsyncMock() + runner.adapters = {Platform.TELEGRAM: mock_adapter} + with patch("gateway.run._hermes_home", hermes_home), patch( + "gateway.run._current_systemd_gateway_service", + return_value="hermes-gateway-default", + create=True, + ): + assert await runner._send_update_notification() is False + + mock_adapter.send.assert_not_awaited() + assert owner_pending.exists() + assert pending_path.exists() + assert not (hermes_home / ".update_exit_code").exists() + + @pytest.mark.asyncio + async def test_deferred_result_without_update_claim_is_not_promoted(self, tmp_path): + runner = _make_runner() + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + owner_pending = hermes_home / ".update_owner_restart_pending.json" + owner_pending.write_text(json.dumps({ + "version": 1, + "exit_code": 0, + "owner_pid": os.getpid() + 1000, + "owner_service": "hermes-gateway-default", + })) + + with patch("gateway.run._hermes_home", hermes_home), patch( + "gateway.run._current_systemd_gateway_service", + return_value="hermes-gateway-default", + create=True, + ): + assert await runner._send_update_notification() is False + + assert owner_pending.exists() + assert not (hermes_home / ".update_exit_code").exists() + @pytest.mark.asyncio async def test_recovers_from_claimed_pending_file(self, tmp_path): """A claimed pending file from a crashed notifier is still deliverable.""" diff --git a/tests/hermes_cli/test_cmd_update.py b/tests/hermes_cli/test_cmd_update.py index e2a546da587c..8ef93dcf9708 100644 --- a/tests/hermes_cli/test_cmd_update.py +++ b/tests/hermes_cli/test_cmd_update.py @@ -67,10 +67,82 @@ def _fake_update_managed_uv(**_kwargs): with patch("hermes_cli.managed_uv.resolve_uv", side_effect=_fake_resolve_uv), \ patch("hermes_cli.managed_uv.ensure_uv", side_effect=_fake_ensure_uv), \ - patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv): + patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv), \ + patch("hermes_cli.main._get_pid_cgroup_path", return_value=None), \ + patch("hermes_cli.main._get_systemd_service_for_pid", return_value=None), \ + patch("hermes_cli.gateway._get_service_pids", return_value=set()), \ + patch("hermes_cli.gateway.find_gateway_pids", return_value=[]), \ + patch("hermes_cli.gateway.find_profile_gateway_processes", return_value=[]): yield +class TestCmdUpdateOwnerFinalizationOrdering: + def test_full_update_defers_owner_until_dashboard_and_external_verifier( + self, monkeypatch + ): + """Drive the full update path, not only finalization helper calls.""" + from hermes_cli import gateway as gateway_cli + from hermes_cli import update_cmd + + owner = ( + "user", + ["systemctl", "--user"], + "hermes-gateway-coding_lead", + ) + events: list[str] = [] + + def detect_owner(): + events.append("detect-owner") + return owner, False + + def inventory(_text, *, process_unit, on_unit_timeout, defer_unit): + events.append(f"inventory:{defer_unit}") + return [defer_unit] if defer_unit else [] + + def dashboard(_failures): + events.append("dashboard") + return True + + def launch(deferred_owner, *, final_exit_code, persist_result): + events.append( + f"verifier:{deferred_owner[2]}:{final_exit_code}:{persist_result}" + ) + return True + + monkeypatch.setattr(update_cmd, "_detect_updater_systemd_gateway_owner", detect_owner) + monkeypatch.setattr(update_cmd, "_for_each_systemd_gateway_unit", inventory) + monkeypatch.setattr(update_cmd, "_finish_dashboard_update_cleanup", dashboard) + monkeypatch.setattr( + update_cmd, "_launch_updater_owner_restart_verifier", launch + ) + monkeypatch.setattr(update_cmd._time, "sleep", lambda _seconds: None) + monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True) + monkeypatch.setattr(gateway_cli, "is_macos", lambda: False) + + with patch("shutil.which", return_value=None), patch( + "subprocess.run" + ) as mock_run, patch( + "hermes_cli.config.get_missing_env_vars", return_value=[] + ), patch( + "hermes_cli.config.get_missing_config_fields", return_value=[] + ), patch( + "hermes_cli.config.check_config_version", return_value=(24, 24) + ): + mock_run.side_effect = _make_run_side_effect( + branch="main", verify_ok=True, commit_count="1" + ) + with pytest.raises(SystemExit) as exc_info: + cmd_update(SimpleNamespace(yes=True)) + + assert exc_info.value.code == 1 + assert events.index("detect-owner") < events.index( + "inventory:hermes-gateway-coding_lead" + ) + assert events.index("dashboard") < events.index( + "verifier:hermes-gateway-coding_lead:0:True" + ) + + class TestCmdUpdateNpmLockfileCache: @staticmethod def _cache_file(hermes_root, project_root): diff --git a/tests/hermes_cli/test_update_fleet_restart_timeout.py b/tests/hermes_cli/test_update_fleet_restart_timeout.py index 2de629208153..7feb7b9588e1 100644 --- a/tests/hermes_cli/test_update_fleet_restart_timeout.py +++ b/tests/hermes_cli/test_update_fleet_restart_timeout.py @@ -24,6 +24,27 @@ def _list_units_stdout(names: list[str]) -> str: class TestFleetRestartTimeoutIsolation: + def test_updater_owning_unit_is_deferred_until_explicit_final_restart(self): + units = [ + "hermes-gateway-default", + "hermes-gateway-coding_lead", + "hermes-gateway-researcher", + ] + restarted: list[str] = [] + + deferred = _for_each_systemd_gateway_unit( + _list_units_stdout(units), + process_unit=restarted.append, + on_unit_timeout=lambda *_: pytest.fail("unexpected timeout"), + defer_unit="hermes-gateway-coding_lead", + ) + + assert restarted == [ + "hermes-gateway-default", + "hermes-gateway-researcher", + ] + assert deferred == ["hermes-gateway-coding_lead"] + def test_timeout_on_middle_unit_continues_remaining_units(self): units = [ "hermes-gateway-xiaomo1", diff --git a/tests/hermes_cli/test_update_owner_finalization.py b/tests/hermes_cli/test_update_owner_finalization.py new file mode 100644 index 000000000000..157d36e7b8f3 --- /dev/null +++ b/tests/hermes_cli/test_update_owner_finalization.py @@ -0,0 +1,579 @@ +"""Ownership-aware updater finalization regressions. + +The gateway that launched ``hermes update --gateway`` may also own the +updater process through its systemd cgroup. Restarting that unit before the +dashboard and result marker are finalized kills the updater mid-flight. +""" + +from __future__ import annotations + +import json +import sys +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from hermes_cli import update_cmd, update_owner_restart + + +def _owner() -> tuple[str, list[str], str]: + return ( + "user", + ["systemctl", "--user"], + "hermes-gateway-coding_lead", + ) + + +def _system_owner() -> tuple[str, list[str], str]: + return ( + "system", + ["systemctl"], + "hermes-gateway-coding_lead", + ) + + +def test_dashboard_precedes_terminal_owner_verifier_without_early_success( + monkeypatch, capsys +): + events: list[str] = [] + + monkeypatch.setattr( + update_cmd, + "_finish_dashboard_update_cleanup", + lambda _failures: events.append("dashboard") or True, + ) + monkeypatch.setattr( + update_cmd, + "_launch_updater_owner_restart_verifier", + lambda _owner, *, final_exit_code, persist_result: events.append( + f"verifier:{final_exit_code}:{persist_result}" + ) + or True, + raising=False, + ) + + ok = update_cmd._finish_update_service_finalization( + [], + gateway_mode=True, + gateway_fleet_restart_incomplete=False, + updater_owner_discovery_failed=False, + deferred_owner=_owner(), + ) + + assert ok is False + assert events == ["dashboard", "verifier:0:True"] + output = capsys.readouterr().out + assert "verification is pending" in output + assert "✓ Update complete!" not in output + + +def test_partial_failure_is_staged_before_owner_verifier(monkeypatch, capsys): + events: list[str] = [] + + monkeypatch.setattr( + update_cmd, + "_finish_dashboard_update_cleanup", + lambda _failures: events.append("dashboard") or False, + ) + monkeypatch.setattr( + update_cmd, + "_launch_updater_owner_restart_verifier", + lambda _owner, *, final_exit_code, persist_result: events.append( + f"verifier:{final_exit_code}:{persist_result}" + ) + or True, + raising=False, + ) + + ok = update_cmd._finish_update_service_finalization( + [], + gateway_mode=True, + gateway_fleet_restart_incomplete=False, + updater_owner_discovery_failed=False, + deferred_owner=_owner(), + ) + + assert ok is False + assert events == ["dashboard", "verifier:1:True"] + assert "verification is pending" in capsys.readouterr().out + + +def test_owner_verifier_launch_failure_persists_error_result(monkeypatch, capsys): + events: list[str] = [] + + monkeypatch.setattr( + update_cmd, "_finish_dashboard_update_cleanup", lambda _failures: True + ) + monkeypatch.setattr( + update_cmd, + "_write_gateway_update_exit_code", + lambda _required, code: events.append(f"status:{code}") or True, + ) + monkeypatch.setattr( + update_cmd, + "_launch_updater_owner_restart_verifier", + lambda _owner, *, final_exit_code, persist_result: events.append( + f"verifier:{final_exit_code}:{persist_result}" + ) + or False, + raising=False, + ) + + ok = update_cmd._finish_update_service_finalization( + [], + gateway_mode=True, + gateway_fleet_restart_incomplete=False, + updater_owner_discovery_failed=False, + deferred_owner=_owner(), + ) + + assert ok is False + assert events == ["verifier:0:True", "status:1"] + assert "Could not launch" in capsys.readouterr().out + + +def test_non_systemd_fallback_persists_final_status_without_verifier(monkeypatch): + events: list[str] = [] + monkeypatch.setattr( + update_cmd, + "_finish_dashboard_update_cleanup", + lambda _failures: events.append("dashboard") or True, + ) + monkeypatch.setattr( + update_cmd, + "_write_gateway_update_exit_code", + lambda required, code: events.append(f"status:{required}:{code}") or True, + ) + monkeypatch.setattr( + update_cmd, + "_launch_updater_owner_restart_verifier", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("non-systemd fallback must not launch verifier") + ), + raising=False, + ) + + assert update_cmd._finish_update_service_finalization( + [], + gateway_mode=True, + gateway_fleet_restart_incomplete=False, + updater_owner_discovery_failed=False, + deferred_owner=None, + ) is True + + assert events == ["dashboard", "status:True:0"] + + +def test_owner_verifier_timeout_covers_after_turn_wait(monkeypatch): + from hermes_cli import gateway as gateway_cli + + monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 1.0) + monkeypatch.setattr(gateway_cli, "_get_restart_exit_wait_budget", lambda: 700.0) + monkeypatch.setattr( + update_cmd, + "_service_restart_sec", + lambda _scope_cmd, _service, *, default: 5.0, + ) + + assert update_cmd._owner_restart_verification_timeout(_owner()) == 825.0 + + +def test_system_owner_timeout_probe_fails_without_prompt_free_privilege(monkeypatch): + commands: list[list[str]] = [] + monkeypatch.setattr(update_cmd.os, "geteuid", lambda: 1000) + monkeypatch.setattr( + update_cmd, + "_service_restart_sec", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("unresolved systemctl command must not run") + ), + ) + + def fake_run(command, **_kwargs): + commands.append(command) + return SimpleNamespace(returncode=1, stdout="", stderr="password required") + + monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + + with pytest.raises(PermissionError, match="prompt-free systemctl"): + update_cmd._owner_restart_verification_timeout(_system_owner()) + + assert commands == [ + ["sudo", "-n", "true"], + [ + "sudo", + "-n", + "systemctl", + "--no-ask-password", + "show", + "hermes-gateway-coding_lead", + "--property=RestartUSec", + "--value", + ], + ] + + +def test_owner_request_is_durable_before_transient_verifier_launch( + tmp_path, monkeypatch +): + monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: tmp_path) + monkeypatch.setattr(update_cmd.shutil, "which", lambda name: "/bin/systemd-run") + monkeypatch.setattr( + update_cmd, + "_owner_restart_verification_timeout", + lambda _owner: 120.0, + raising=False, + ) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: { + "active_state": "active", + "sub_state": "running", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "on-failure", + }, + ) + monkeypatch.setattr(update_cmd.secrets, "token_hex", lambda _size: "a" * 32) + launches: list[list[str]] = [] + + def fake_run(args, *unused_args, **unused_kwargs): + pending = json.loads( + (tmp_path / update_owner_restart.OWNER_RESTART_PENDING_FILE).read_text() + ) + assert pending["version"] == 2 + assert pending["old_state"]["main_pid"] == 4321 + assert pending["generation_key"] == "exec_start" + assert not (tmp_path / ".update_exit_code").exists() + launches.append(args) + return MagicMock(returncode=0, stdout="", stderr="") + + with patch.object(update_cmd.subprocess, "run", side_effect=fake_run), patch.object( + update_cmd.os, + "kill", + side_effect=AssertionError("updater cgroup must not own the terminal signal"), + ): + assert update_cmd._launch_updater_owner_restart_verifier( + _owner(), final_exit_code=0, persist_result=True + ) is True + + command = launches[0] + assert command[0:3] == [ + "/bin/systemd-run", + "--user", + "--no-ask-password", + ] + assert "--collect" in command + assert any(part.startswith("--unit=hermes-update-owner-verify-") for part in command) + assert command[-6:] == [ + sys.executable, + "-m", + "hermes_cli.update_owner_restart", + "--home", + str(tmp_path), + f"--nonce={'a' * 32}", + ] + + +def test_root_system_owner_verifier_launch_never_prompts(tmp_path, monkeypatch): + monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: tmp_path) + monkeypatch.setattr(update_cmd.shutil, "which", lambda name: "/bin/systemd-run") + monkeypatch.setattr(update_cmd.os, "geteuid", lambda: 0) + monkeypatch.setattr(update_cmd.os, "getegid", lambda: 0) + monkeypatch.setattr( + update_cmd, + "_owner_restart_verification_timeout", + lambda _owner: 120.0, + ) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: { + "active_state": "active", + "sub_state": "running", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "on-failure", + }, + ) + monkeypatch.setattr(update_cmd.secrets, "token_hex", lambda _size: "b" * 32) + launches: list[list[str]] = [] + + def fake_run(args, *unused_args, **unused_kwargs): + launches.append(args) + return MagicMock(returncode=0, stdout="", stderr="") + + with patch.object(update_cmd.subprocess, "run", side_effect=fake_run): + assert update_cmd._launch_updater_owner_restart_verifier( + _system_owner(), final_exit_code=0, persist_result=True + ) is True + + assert len(launches) == 1 + assert launches[0][0:2] == ["/bin/systemd-run", "--no-ask-password"] + assert "--user" not in launches[0] + assert "--property=User=0" in launches[0] + assert "--property=Group=0" in launches[0] + + +def test_system_owner_verifier_refuses_prompting_privilege_path(tmp_path, monkeypatch): + monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: tmp_path) + monkeypatch.setattr(update_cmd.shutil, "which", lambda name: "/bin/systemd-run") + monkeypatch.setattr(update_cmd.os, "geteuid", lambda: 1000) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda *_args: (_ for _ in ()).throw( + AssertionError("privilege must resolve before reading owner state") + ), + ) + commands: list[list[str]] = [] + + def fake_run(args, *unused_args, **unused_kwargs): + commands.append(args) + if args in ( + ["sudo", "-n", "true"], + [ + "sudo", + "-n", + "/bin/systemd-run", + "--no-ask-password", + "--version", + ], + ): + return MagicMock(returncode=1, stdout="", stderr="password required") + raise AssertionError(f"unexpected prompting or launch command: {args}") + + with patch.object(update_cmd.subprocess, "run", side_effect=fake_run), patch.object( + update_cmd.os, + "kill", + side_effect=AssertionError("owner must remain untouched"), + ): + assert update_cmd._launch_updater_owner_restart_verifier( + _system_owner(), final_exit_code=0, persist_result=True + ) is False + + assert commands == [ + ["sudo", "-n", "true"], + [ + "sudo", + "-n", + "/bin/systemd-run", + "--no-ask-password", + "--version", + ], + ] + assert not ( + tmp_path / update_owner_restart.OWNER_RESTART_PENDING_FILE + ).exists() + + +def test_system_owner_verifier_uses_sudo_n_when_capability_exists( + tmp_path, monkeypatch +): + monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: tmp_path) + monkeypatch.setattr(update_cmd.shutil, "which", lambda name: "/bin/systemd-run") + monkeypatch.setattr(update_cmd.os, "geteuid", lambda: 1000) + monkeypatch.setattr(update_cmd.os, "getegid", lambda: 1000) + monkeypatch.setattr( + update_cmd, + "_owner_restart_verification_timeout", + lambda _owner: 120.0, + ) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: { + "active_state": "active", + "sub_state": "running", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "on-failure", + }, + ) + monkeypatch.setattr(update_cmd.secrets, "token_hex", lambda _size: "d" * 32) + commands: list[list[str]] = [] + + def fake_run(args, *unused_args, **unused_kwargs): + commands.append(args) + if args == ["sudo", "-n", "true"]: + return MagicMock(returncode=0, stdout="", stderr="") + assert args[0:4] == [ + "sudo", + "-n", + "/bin/systemd-run", + "--no-ask-password", + ] + return MagicMock(returncode=0, stdout="", stderr="") + + with patch.object(update_cmd.subprocess, "run", side_effect=fake_run): + assert update_cmd._launch_updater_owner_restart_verifier( + _system_owner(), final_exit_code=0, persist_result=True + ) is True + + assert len(commands) == 2 + launch = commands[1] + assert "--user" not in launch + assert "--property=User=1000" in launch + assert "--property=Group=1000" in launch + + +def test_system_owner_targeted_sudo_probe_can_resolve_prompt_free_command( + monkeypatch, +): + calls = [] + + def fake_run(command, **kwargs): + calls.append((command, kwargs)) + return SimpleNamespace( + args=command, + returncode=0 if command[-1] == "--version" else 1, + stdout="", + stderr="", + ) + + monkeypatch.setattr(update_cmd.os, "geteuid", lambda: 1000) + monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + command = ["/bin/systemd-run", "--no-ask-password"] + + assert update_cmd._resolve_noninteractive_systemd_command( + "system", + command, + targeted_probe=[*command, "--version"], + ) == ["sudo", "-n", *command] + assert [call[0] for call in calls] == [ + ["sudo", "-n", "true"], + ["sudo", "-n", *command, "--version"], + ] + assert all(call[1]["timeout"] == 5 for call in calls) + + +def test_owner_is_not_restarted_when_request_cannot_be_written(monkeypatch): + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: { + "active_state": "active", + "sub_state": "running", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "on-failure", + }, + ) + monkeypatch.setattr( + update_owner_restart, + "prepare_owner_restart_request", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("read-only home")), + ) + monkeypatch.setattr( + update_cmd, + "_owner_restart_verification_timeout", + lambda _owner: 120.0, + ) + with patch.object(update_cmd.subprocess, "run") as run: + assert update_cmd._launch_updater_owner_restart_verifier( + _owner(), final_exit_code=0, persist_result=True + ) is False + + run.assert_not_called() + + +def test_owner_detection_uses_pid_cgroup_scope(monkeypatch): + live = SimpleNamespace( + _get_pid_cgroup_path=lambda _pid: ( + "/user.slice/user-1000.slice/hermes-gateway-coding_lead.service" + ), + _get_systemd_service_for_pid=lambda _pid: ( + "hermes-gateway-coding_lead.service" + ), + _extract_scope_from_cgroup=lambda _path: "user", + ) + monkeypatch.setattr(update_cmd, "_m", lambda: live) + + owner, failed = update_cmd._detect_updater_systemd_gateway_owner() + + assert owner == _owner() + assert failed is False + + +def test_nested_gateway_cgroup_resolves_owner_component(monkeypatch): + live = SimpleNamespace( + _get_pid_cgroup_path=lambda _pid: ( + "/user.slice/user-1000.slice/user@1000.service/app.slice/" + "hermes-gateway-coding_lead.service/worker.scope" + ), + _get_systemd_service_for_pid=lambda _pid: None, + _extract_scope_from_cgroup=lambda _path: "user", + ) + monkeypatch.setattr(update_cmd, "_m", lambda: live) + + assert update_cmd._detect_updater_systemd_gateway_owner() == (_owner(), False) + + +def test_unknown_non_systemd_ownership_is_not_a_failure(monkeypatch): + live = SimpleNamespace( + _get_pid_cgroup_path=lambda _pid: None, + _get_systemd_service_for_pid=lambda _pid: None, + _extract_scope_from_cgroup=lambda _path: None, + ) + monkeypatch.setattr(update_cmd, "_m", lambda: live) + + assert update_cmd._detect_updater_systemd_gateway_owner() == (None, False) + + +def test_malformed_gateway_cgroup_that_cannot_resolve_owner_is_a_failure(monkeypatch): + live = SimpleNamespace( + _get_pid_cgroup_path=lambda _pid: ( + "/user.slice/user-1000.slice/hermes-gateway-coding_lead.service.extra" + ), + _get_systemd_service_for_pid=lambda _pid: None, + _extract_scope_from_cgroup=lambda _path: "user", + ) + monkeypatch.setattr(update_cmd, "_m", lambda: live) + + assert update_cmd._detect_updater_systemd_gateway_owner() == (None, True) + + +def test_transient_verifier_launch_failure_clears_pending_request( + tmp_path, monkeypatch +): + monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: tmp_path) + monkeypatch.setattr(update_cmd.shutil, "which", lambda _name: "/bin/systemd-run") + monkeypatch.setattr( + update_cmd, + "_owner_restart_verification_timeout", + lambda _owner: 120.0, + raising=False, + ) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: { + "active_state": "active", + "sub_state": "running", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "on-failure", + }, + ) + monkeypatch.setattr(update_cmd.secrets, "token_hex", lambda _size: "c" * 32) + monkeypatch.setattr( + update_cmd.subprocess, + "run", + lambda *_args, **_kwargs: MagicMock( + returncode=1, stdout="", stderr="permission denied" + ), + ) + + assert update_cmd._launch_updater_owner_restart_verifier( + _owner(), final_exit_code=0, persist_result=True + ) is False + assert not ( + tmp_path / update_owner_restart.OWNER_RESTART_PENDING_FILE + ).exists() diff --git a/tests/hermes_cli/test_update_owner_restart_verifier.py b/tests/hermes_cli/test_update_owner_restart_verifier.py new file mode 100644 index 000000000000..0930edc8e19d --- /dev/null +++ b/tests/hermes_cli/test_update_owner_restart_verifier.py @@ -0,0 +1,420 @@ +"""Out-of-cgroup updater-owner restart verification regressions.""" + +from __future__ import annotations + +import json +import signal +from pathlib import Path + +import pytest + +from hermes_cli import update_owner_restart + + +SERVICE = "hermes-gateway-coding_lead" +OLD_STATE = { + "active_state": "active", + "sub_state": "running", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "on-failure", +} +NEW_STATE = { + "active_state": "active", + "sub_state": "running", + "main_pid": 9876, + "exec_start": 201, + "active_enter": 202, + "restart": "on-failure", +} + + +def _prepare( + home: Path, + *, + restart: str = "on-failure", + timeout_seconds: float = 0.03, + nonce: str = "a" * 32, +) -> dict: + old_state = {**OLD_STATE, "restart": restart} + return update_owner_restart.prepare_owner_restart_request( + home, + scope="user", + service=SERVICE, + old_state=old_state, + final_exit_code=0, + timeout_seconds=timeout_seconds, + nonce=nonce, + ) + + +def test_request_accepts_bounded_full_gateway_exit_wait(tmp_path): + request = _prepare(tmp_path, timeout_seconds=1800.0) + + assert request["timeout_seconds"] == 1800.0 + + with pytest.raises(ValueError, match="owner restart timeout"): + _prepare(tmp_path, timeout_seconds=1800.1) + + +def _systemd_state_result(*, returncode: int = 0): + return type( + "Result", + (), + { + "returncode": returncode, + "stdout": ( + "ActiveState=active\n" + "SubState=running\n" + "MainPID=4321\n" + "ExecMainStartTimestampMonotonic=101\n" + "ActiveEnterTimestampMonotonic=102\n" + "Restart=on-failure\n" + ) + if returncode == 0 + else "", + "stderr": "" if returncode == 0 else "permission denied", + }, + )() + + +def test_root_system_scope_state_read_disables_password_prompts(monkeypatch): + commands: list[list[str]] = [] + monkeypatch.setattr(update_owner_restart.os, "geteuid", lambda: 0) + monkeypatch.setattr( + update_owner_restart.subprocess, + "run", + lambda args, **_kwargs: commands.append(args) or _systemd_state_result(), + ) + + assert update_owner_restart.read_systemd_service_state("system", SERVICE) == OLD_STATE + assert commands[0][0:3] == [ + "systemctl", + "--no-ask-password", + "show", + ] + + +def test_nonroot_system_scope_state_read_uses_sudo_n(monkeypatch): + commands: list[list[str]] = [] + monkeypatch.setattr(update_owner_restart.os, "geteuid", lambda: 1000) + + def fake_run(args, **_kwargs): + commands.append(args) + if args == ["sudo", "-n", "true"]: + return _systemd_state_result() + assert args[0:5] == [ + "sudo", + "-n", + "systemctl", + "--no-ask-password", + "show", + ] + return _systemd_state_result() + + monkeypatch.setattr(update_owner_restart.subprocess, "run", fake_run) + + assert update_owner_restart.read_systemd_service_state("system", SERVICE) == OLD_STATE + assert commands[0] == ["sudo", "-n", "true"] + assert commands[1][0:5] == [ + "sudo", + "-n", + "systemctl", + "--no-ask-password", + "show", + ] + + +def test_targeted_system_scope_state_probe_is_reused(monkeypatch): + commands: list[list[str]] = [] + monkeypatch.setattr(update_owner_restart.os, "geteuid", lambda: 1000) + + def fake_run(args, **_kwargs): + commands.append(args) + if args == ["sudo", "-n", "true"]: + return _systemd_state_result(returncode=1) + assert args[0:5] == [ + "sudo", + "-n", + "systemctl", + "--no-ask-password", + "show", + ] + return _systemd_state_result() + + monkeypatch.setattr(update_owner_restart.subprocess, "run", fake_run) + + assert update_owner_restart.read_systemd_service_state("system", SERVICE) == OLD_STATE + assert len(commands) == 2 + + +def _result(home: Path) -> dict: + return json.loads( + (home / update_owner_restart.OWNER_RESTART_RESULT_FILE).read_text() + ) + + +def test_signal_accepted_without_owner_transition_persists_failure( + tmp_path, monkeypatch +): + request = _prepare(tmp_path) + signals: list[tuple[int, signal.Signals]] = [] + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: dict(OLD_STATE), + ) + monkeypatch.setattr( + update_owner_restart.os, + "kill", + lambda pid, sig: signals.append((pid, sig)), + ) + + assert ( + update_owner_restart.verify_owner_restart( + tmp_path, request["nonce"], poll_interval=0.001 + ) + == 1 + ) + + assert signals == [(4321, signal.SIGUSR1)] + assert (tmp_path / ".update_exit_code").read_text() == "1" + assert _result(tmp_path)["verified"] is False + assert "timed out" in _result(tmp_path)["reason"] + assert "✓ Update complete!" not in ( + tmp_path / ".update_output.txt" + ).read_text() + + +def test_changed_pid_and_generation_require_matching_health_ack(tmp_path, monkeypatch): + request = _prepare(tmp_path) + states = iter([dict(OLD_STATE), dict(NEW_STATE)]) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: next(states, dict(NEW_STATE)), + ) + monkeypatch.setattr(update_owner_restart.os, "kill", lambda _pid, _sig: None) + assert update_owner_restart.acknowledge_owner_restart_ready( + tmp_path, + current_service=SERVICE, + current_pid=9876, + now_ns=request["requested_at_ns"] + 1, + ) + + assert ( + update_owner_restart.verify_owner_restart( + tmp_path, request["nonce"], poll_interval=0.001 + ) + == 0 + ) + + result = _result(tmp_path) + assert result["verified"] is True + assert result["observed_state"]["main_pid"] == 9876 + assert result["observed_state"]["exec_start"] == 201 + assert (tmp_path / ".update_exit_code").read_text() == "0" + assert "✓ Update complete!" in (tmp_path / ".update_output.txt").read_text() + + +def test_restart_no_exit_is_actionable_without_automatic_start(tmp_path, monkeypatch): + request = _prepare(tmp_path, restart="no") + old_no_restart = {**OLD_STATE, "restart": "no"} + deactivating = { + "active_state": "deactivating", + "sub_state": "stop-sigterm", + "main_pid": 4321, + "exec_start": 101, + "active_enter": 102, + "restart": "no", + } + deactivating_without_pid = {**deactivating, "main_pid": 0} + inactive = { + "active_state": "inactive", + "sub_state": "dead", + "main_pid": 0, + "exec_start": 101, + "active_enter": 102, + "restart": "no", + } + states = iter( + [old_no_restart, deactivating, deactivating_without_pid, inactive] + ) + + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: next(states, dict(inactive)), + ) + monkeypatch.setattr(update_owner_restart.os, "kill", lambda _pid, _sig: None) + monkeypatch.setattr( + update_owner_restart.subprocess, + "run", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("Restart=no must never be auto-started") + ), + ) + + assert ( + update_owner_restart.verify_owner_restart( + tmp_path, request["nonce"], poll_interval=0.001 + ) + == 1 + ) + + result = _result(tmp_path) + assert result["verified"] is False + assert result["exit_code"] == 1 + assert result["reason"] == "owner exited and Restart=no disables automatic restart" + assert result["observed_state"]["active_state"] == "inactive" + assert result["observed_state"]["main_pid"] == 0 + assert result["message"] == ( + "✗ Update finalization incomplete: updater-owning service " + f"{SERVICE} exited and has Restart=no; Hermes did not auto-start it.\n" + " Restart it manually to load the updated code:\n" + f" systemctl --user start {SERVICE}\n" + " Then verify:\n" + f" systemctl --user status {SERVICE}" + ) + assert (tmp_path / ".update_exit_code").read_text() == "1" + + +def test_restart_no_prearmed_transition_cannot_publish_success(tmp_path, monkeypatch): + request = _prepare(tmp_path, restart="no") + new_no_restart = {**NEW_STATE, "restart": "no"} + assert update_owner_restart.acknowledge_owner_restart_ready( + tmp_path, + current_service=SERVICE, + current_pid=9876, + now_ns=request["requested_at_ns"] + 1, + ) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: dict(new_no_restart), + ) + monkeypatch.setattr( + update_owner_restart.os, + "kill", + lambda *_args: (_ for _ in ()).throw( + AssertionError("prearmed transition must not signal again") + ), + ) + + assert update_owner_restart.verify_owner_restart(tmp_path, request["nonce"]) == 1 + result = _result(tmp_path) + assert result["verified"] is False + assert result["exit_code"] == 1 + assert result["reason"] == "owner exited and Restart=no disables automatic restart" + assert "systemctl --user start" in result["message"] + + +def test_changed_owner_without_health_ack_times_out_as_failure(tmp_path, monkeypatch): + request = _prepare(tmp_path) + states = iter([dict(OLD_STATE), dict(NEW_STATE)]) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: next(states, dict(NEW_STATE)), + ) + monkeypatch.setattr(update_owner_restart.os, "kill", lambda _pid, _sig: None) + + assert ( + update_owner_restart.verify_owner_restart( + tmp_path, request["nonce"], poll_interval=0.001 + ) + == 1 + ) + assert _result(tmp_path)["verified"] is False + assert "readiness acknowledgement" in _result(tmp_path)["reason"] + + +def test_stale_and_foreign_requests_are_not_acknowledged(tmp_path): + stale = _prepare(tmp_path, timeout_seconds=0.01) + assert not update_owner_restart.acknowledge_owner_restart_ready( + tmp_path, + current_service=SERVICE, + current_pid=9876, + now_ns=stale["deadline_ns"] + 1, + ) + assert not (tmp_path / update_owner_restart.OWNER_RESTART_ACK_FILE).exists() + + foreign = _prepare(tmp_path, nonce="b" * 32) + assert not update_owner_restart.acknowledge_owner_restart_ready( + tmp_path, + current_service="hermes-gateway-other", + current_pid=9876, + now_ns=foreign["requested_at_ns"] + 1, + ) + assert not (tmp_path / update_owner_restart.OWNER_RESTART_ACK_FILE).exists() + + +def test_foreign_nonce_is_rejected_without_signalling(tmp_path, monkeypatch): + _prepare(tmp_path) + monkeypatch.setattr( + update_owner_restart.os, + "kill", + lambda *_args: (_ for _ in ()).throw(AssertionError("must not signal")), + ) + + assert update_owner_restart.verify_owner_restart(tmp_path, "f" * 32) == 1 + assert not (tmp_path / ".update_exit_code").exists() + + +def test_duplicate_verifier_invocation_is_idempotent(tmp_path, monkeypatch): + request = _prepare(tmp_path) + states = iter([dict(OLD_STATE), dict(NEW_STATE)]) + signals: list[tuple[int, signal.Signals]] = [] + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: next(states, dict(NEW_STATE)), + ) + monkeypatch.setattr( + update_owner_restart.os, + "kill", + lambda pid, sig: signals.append((pid, sig)), + ) + assert update_owner_restart.acknowledge_owner_restart_ready( + tmp_path, + current_service=SERVICE, + current_pid=9876, + now_ns=request["requested_at_ns"] + 1, + ) + + assert update_owner_restart.verify_owner_restart(tmp_path, request["nonce"]) == 0 + first_output = (tmp_path / ".update_output.txt").read_text() + assert update_owner_restart.verify_owner_restart(tmp_path, request["nonce"]) == 0 + + assert signals == [(4321, signal.SIGUSR1)] + assert (tmp_path / ".update_output.txt").read_text() == first_output + + +def test_atomic_result_remains_terminal_when_legacy_marker_write_fails( + tmp_path, monkeypatch +): + request = _prepare(tmp_path) + states = iter([dict(OLD_STATE), dict(NEW_STATE)]) + monkeypatch.setattr( + update_owner_restart, + "read_systemd_service_state", + lambda _scope, _service: next(states, dict(NEW_STATE)), + ) + monkeypatch.setattr(update_owner_restart.os, "kill", lambda _pid, _sig: None) + assert update_owner_restart.acknowledge_owner_restart_ready( + tmp_path, + current_service=SERVICE, + current_pid=9876, + now_ns=request["requested_at_ns"] + 1, + ) + monkeypatch.setattr( + update_owner_restart, + "atomic_write_text", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("read-only marker")), + ) + + assert update_owner_restart.verify_owner_restart(tmp_path, request["nonce"]) == 0 + assert update_owner_restart.verify_owner_restart(tmp_path, request["nonce"]) == 0 + assert update_owner_restart.read_owner_restart_result_exit_code(tmp_path) == 0 + assert not (tmp_path / ".update_exit_code").exists() diff --git a/tests/hermes_cli/test_update_stale_dashboard.py b/tests/hermes_cli/test_update_stale_dashboard.py index e61fe8acbb7d..bf01b7d1d29d 100644 --- a/tests/hermes_cli/test_update_stale_dashboard.py +++ b/tests/hermes_cli/test_update_stale_dashboard.py @@ -238,11 +238,216 @@ def test_all_failed_stops_do_not_claim_the_dashboard_was_stopped(self, capsys): "hermes_cli.main._kill_stale_dashboard_processes", return_value={"matched": [12345], "killed": [], "failed": [(12345, "denied")], "unrecovered": []}, - ): - _finish_dashboard_update_cleanup([]) + ), patch("hermes_cli.main._snapshot_managed_dashboard_service", return_value=None): + assert _finish_dashboard_update_cleanup([]) is False assert "stopped during update" not in capsys.readouterr().out + def test_managed_restart_is_verified_after_cleanup(self): + before = { + "scope": ("--user",), + "unit": "hermes-dashboard.service", + "active": True, + "main_pid": 12345, + "started": 10, + "runtime": ("dashboard", "127.0.0.1", 9119), + } + with patch( + "hermes_cli.main._snapshot_managed_dashboard_service", + return_value=before, + ), patch( + "hermes_cli.main._kill_stale_dashboard_processes", + return_value={"matched": [], "killed": [], "failed": [], "unrecovered": []}, + ), patch( + "hermes_cli.main._verify_managed_dashboard_restart", + return_value=True, + ) as verify: + assert _finish_dashboard_update_cleanup([]) is True + + verify.assert_called_once_with(before) + + def test_managed_restart_verification_failure_is_non_success(self, capsys): + before = { + "scope": ("--user",), + "unit": "hermes-dashboard.service", + "active": True, + "main_pid": 12345, + "started": 10, + "runtime": ("dashboard", "127.0.0.1", 9119), + } + with patch( + "hermes_cli.main._snapshot_managed_dashboard_service", + return_value=before, + ), patch( + "hermes_cli.main._kill_stale_dashboard_processes", + return_value={"matched": [], "killed": [], "failed": [], "unrecovered": []}, + ), patch( + "hermes_cli.main._verify_managed_dashboard_restart", + return_value=False, + ): + assert _finish_dashboard_update_cleanup([]) is False + + assert "could not be verified" in capsys.readouterr().out + + def test_custom_systemd_backend_restart_is_also_verified(self): + before = { + "scope": (), + "unit": "hermes-serve.service", + "active": True, + "main_pid": 4444, + "started": 10, + "runtime": None, + } + stop_result = { + "matched": [4444], + "killed": [4444], + "failed": [], + "unrecovered": [], + "restarted_service_snapshots": [before], + "unverified_services": [], + } + with patch( + "hermes_cli.main._snapshot_managed_dashboard_service", + return_value=None, + ), patch( + "hermes_cli.main._kill_stale_dashboard_processes", + return_value=stop_result, + ), patch( + "hermes_cli.main._verify_managed_dashboard_restart", + return_value=False, + ) as verify: + assert _finish_dashboard_update_cleanup([]) is False + + verify.assert_called_once_with(before) + + def test_unverified_custom_systemd_backend_is_non_success(self, capsys): + stop_result = { + "matched": [4444], + "killed": [4444], + "failed": [], + "unrecovered": [], + "restarted_service_snapshots": [], + "unverified_services": ["hermes-serve.service"], + } + with patch( + "hermes_cli.main._snapshot_managed_dashboard_service", + return_value=None, + ), patch( + "hermes_cli.main._kill_stale_dashboard_processes", + return_value=stop_result, + ): + assert _finish_dashboard_update_cleanup([]) is False + + assert "could not be verified" in capsys.readouterr().out + + +class TestManagedDashboardRestartVerification: + def _before(self): + return { + "scope": ("--user",), + "unit": "hermes-dashboard.service", + "active": True, + "main_pid": 12345, + "started": 10, + "runtime": ("dashboard", "127.0.0.1", 9119), + } + + def test_active_new_process_and_healthy_endpoint_are_required(self): + live = sys.modules["hermes_cli.main"] + after = {"active": True, "main_pid": 23456, "started": 20} + with patch.object( + live, "_read_managed_dashboard_service_state", return_value=after + ), patch.object(live, "_dashboard_healthcheck", return_value=True) as health: + assert live._verify_managed_dashboard_restart( + self._before(), timeout=0 + ) is True + + health.assert_called_once_with(("dashboard", "127.0.0.1", 9119)) + + def test_same_pid_and_start_time_are_not_a_verified_restart(self): + live = sys.modules["hermes_cli.main"] + unchanged = {"active": True, "main_pid": 12345, "started": 10} + with patch.object( + live, "_read_managed_dashboard_service_state", return_value=unchanged + ), patch.object(live, "_dashboard_healthcheck") as health: + assert live._verify_managed_dashboard_restart( + self._before(), timeout=0 + ) is False + + health.assert_not_called() + + def test_unhealthy_new_process_is_not_success(self): + live = sys.modules["hermes_cli.main"] + after = {"active": True, "main_pid": 23456, "started": 20} + with patch.object( + live, "_read_managed_dashboard_service_state", return_value=after + ), patch.object(live, "_dashboard_healthcheck", return_value=False): + assert live._verify_managed_dashboard_restart( + self._before(), timeout=0 + ) is False + + def test_enabled_but_inactive_unit_is_not_scheduled_for_restart(self): + live = sys.modules["hermes_cli.main"] + + def fake_run(args, *unused_args, **unused_kwargs): + if args == [ + "systemctl", "--user", "list-unit-files", + "hermes-dashboard.service", "--no-legend", "--no-pager", + ]: + return MagicMock( + returncode=0, + stdout="hermes-dashboard.service enabled enabled\n", + stderr="", + ) + if args == [ + "systemctl", "--user", "is-enabled", "hermes-dashboard.service", + ]: + return MagicMock(returncode=0, stdout="enabled\n", stderr="") + if args == [ + "systemctl", "list-unit-files", "hermes-dashboard.service", + "--no-legend", "--no-pager", + ]: + return MagicMock(returncode=0, stdout="", stderr="") + raise AssertionError(f"unexpected subprocess.run call: {args}") + + with patch.object(live.subprocess, "run", side_effect=fake_run), patch.object( + live, + "_read_managed_dashboard_service_state", + return_value={"active": False, "main_pid": 0, "started": 0}, + ): + assert live._snapshot_managed_dashboard_service() is None + + def test_disabled_inactive_unit_is_not_scheduled_for_restart(self): + live = sys.modules["hermes_cli.main"] + + def fake_run(args, *unused_args, **unused_kwargs): + if args == [ + "systemctl", "--user", "list-unit-files", + "hermes-dashboard.service", "--no-legend", "--no-pager", + ]: + return MagicMock( + returncode=0, + stdout="hermes-dashboard.service disabled disabled\n", + stderr="", + ) + if args == [ + "systemctl", "--user", "is-enabled", "hermes-dashboard.service", + ]: + return MagicMock(returncode=1, stdout="disabled\n", stderr="") + if args == [ + "systemctl", "list-unit-files", "hermes-dashboard.service", + "--no-legend", "--no-pager", + ]: + return MagicMock(returncode=0, stdout="", stderr="") + raise AssertionError(f"unexpected subprocess.run call: {args}") + + with patch.object(live.subprocess, "run", side_effect=fake_run), patch.object( + live, + "_read_managed_dashboard_service_state", + return_value={"active": False, "main_pid": 0, "started": 0}, + ): + assert live._snapshot_managed_dashboard_service() is None + class TestWindowsWmicEncoding: """Regression tests for #17049 — the Windows wmic branch must not crash diff --git a/tests/hermes_cli/test_update_yes_flag.py b/tests/hermes_cli/test_update_yes_flag.py index 699d57a97166..fb3943f7bd7a 100644 --- a/tests/hermes_cli/test_update_yes_flag.py +++ b/tests/hermes_cli/test_update_yes_flag.py @@ -12,6 +12,8 @@ from types import SimpleNamespace from unittest.mock import patch +import pytest + from hermes_cli.main import cmd_update @@ -47,6 +49,19 @@ def side_effect(cmd, **kwargs): return side_effect +@pytest.fixture(autouse=True) +def _isolate_update_process_discovery(): + """Keep unit tests from discovering the real worker/gateway cgroup.""" + with patch("hermes_cli.main._get_pid_cgroup_path", return_value=None), patch( + "hermes_cli.main._get_systemd_service_for_pid", return_value=None + ), patch("hermes_cli.gateway._get_service_pids", return_value=set()), patch( + "hermes_cli.gateway.find_gateway_pids", return_value=[] + ), patch( + "hermes_cli.gateway.find_profile_gateway_processes", return_value=[] + ): + yield + + class TestUpdateYesConfigMigration: """--yes auto-answers the config-migration prompt and skips API-key prompts."""