From 27cd0ff4e8b8045d47d6f2c307e6fe7d9352aaf0 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Thu, 27 Aug 2026 13:51:30 -0300 Subject: [PATCH] fix(update): keep serve-unit recovery identity scope-qualified MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review on #96235: discovery distinguished `(scope, unit)`, but the skip payload and the reported outcomes reduced that to the bare service name. `user/hermes-serve.service` and `system/hermes-serve.service` are two different processes, so a single unqualified token could suppress recovery of both: if the user-scope unit was already settled when the restart phase aborted, the stale system-scope unit was never restarted and nothing downstream reported it. Scope now travels with the unit end to end: - the in-process systemd loop records a scope-qualified twin of `restarted_services` (`restarted_scoped_units`) while the bare-name list keeps its existing vocabulary for the fleet probe and the receipt; - the recovery payload carries `{"scope", "unit"}` objects, and the child keys discovery, skips, outcomes and accounting by `(scope, base)`; - `verified` / `failed` — and therefore the receipt and the completion predicate — report `user/hermes-serve`, never a bare name; - an entry with no scope (a payload written by a pre-update interpreter) stays unqualified and is read as scope-agnostic, and an unrecognized scope drops the skip rather than honouring it: dropping a skip can only cost one more restart-and-verify, honouring an unreadable one can leave a stale generation running. Also from review: the survivor probe compared PIDs alone while the plan discarded the process incarnation, so a new serve that reused the planned PID read as the pre-update survivor. The inventory now records the ledger's `create_time` in the serve/dashboard runtime detail and the probe compares `(pid, create_time)`, still failing closed when either side has none. Finally, abort recovery moves out of the update monolith into `hermes_cli/update_abort_recovery.py` (417 lines) with `update_cmd` re-exporting the names `hermes_cli.main` and the update flow address. `update_cmd.py` ends up 75 lines smaller than the PR's base commit instead of 249 lines larger. Tests: dual-scope same-name regressions in both directions, proof that no systemctl verb reaches an already-settled scope, per-scope outcomes, the legacy unqualified shape, the qualified payload shape, scope-qualified completion accounting, PID-reuse vs. same-incarnation survivors, and the inventory carrying `create_time`. Refs #92145 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01YQ9oCBKgAMHSGG8CLEHLMC --- hermes_cli/update_abort_recovery.py | 423 ++++++++++++++++++ hermes_cli/update_cmd.py | 380 ++-------------- hermes_cli/update_inventory.py | 5 + hermes_cli/update_restart_recovery.py | 146 ++++-- .../test_serve_runtime_inventory.py | 19 + .../test_update_restart_recovery.py | 5 +- .../test_update_serve_generation_recovery.py | 308 +++++++++++-- 7 files changed, 875 insertions(+), 411 deletions(-) create mode 100644 hermes_cli/update_abort_recovery.py diff --git a/hermes_cli/update_abort_recovery.py b/hermes_cli/update_abort_recovery.py new file mode 100644 index 0000000000..a4219db82d --- /dev/null +++ b/hermes_cli/update_abort_recovery.py @@ -0,0 +1,423 @@ +"""Fresh-process recovery after the update's in-process restart phase aborts. + +``hermes update`` performs its fleet restart in the interpreter that started +before ``git pull``. When that phase raises — the module graph it is running +no longer matches the checkout on disk — this module owns everything that +happens next: which runtimes a clean child may relaunch, what counts as proof +that a relaunch actually replaced the old generation, which pre-update +processes are still alive, and whether any of that adds up to a recovery that +may call itself complete (#92145). + +It is deliberately a separate owner from ``update_cmd``: the abort path has +its own vocabulary (``verified`` / ``relaunch_attempted`` / ``failed``, serve +units, survivors) and its own fail-closed contract, and the update monolith +must not grow another authority surface (review on #96235). The child process +itself lives in :mod:`hermes_cli.update_restart_recovery`. +""" + +from __future__ import annotations + +import json +import logging +import os +import shutil +import subprocess +import sys + +logger = logging.getLogger(__name__) + + +def _serve_unit_recovery_available() -> bool: + """Can a fresh process restart ``hermes-serve*`` units on this host?""" + return sys.platform == "linux" and bool(shutil.which("systemctl")) + + +def _surviving_pre_update_serve_runtimes(plan) -> list[dict]: + """Pre-update serve/dashboard runtimes that are STILL the same process. + + Identity is the process incarnation ``(pid, create_time)``, never the PID + alone. ``ledger_entries()`` re-verifies that pair on every read and prunes + anything that is gone, so a live entry is a real process — but the numeric + PID can be reused, and a serve that was restarted correctly can come back + on the same number. Comparing PIDs alone would then report the successor + as the pre-update survivor and keep the update incomplete forever + (#92145 review). + + Anything still here after the recovery pass is a live runtime on the + pre-update code generation, which is precisely the unsafe state #92145 + reports. Fail closed on missing evidence: an unreadable ledger, or an + incarnation neither side can produce, counts the runtime as surviving. + """ + planned: dict[int, dict] = {} + try: + for runtime in getattr(plan, "runtimes", ()) or (): + if getattr(runtime, "kind", None) not in ("serve", "dashboard"): + continue + pid = getattr(runtime, "pid", None) + if not isinstance(pid, int) or pid <= 0: + continue + detail = getattr(runtime, "detail", None) + created = detail.get("create_time") if isinstance(detail, dict) else None + planned[pid] = { + "pid": pid, + "kind": str(getattr(runtime, "kind", "")), + "profile": str(getattr(runtime, "profile", "")), + "supervisor": str(getattr(runtime, "supervisor", "")), + "_create_time": created if isinstance(created, (int, float)) else None, + } + except Exception as exc: + logger.debug("Could not read planned serve runtimes: %s", exc) + return [] + if not planned: + return [] + try: + from hermes_cli.process_identity import ledger_entries + + live: dict[int, float | None] = {} + for entry in ledger_entries(): + if entry.get("purpose") not in ("serve", "dashboard"): + continue + pid = entry.get("pid") + if not isinstance(pid, int): + continue + created = entry.get("create_time") + live[pid] = created if isinstance(created, (int, float)) else None + except Exception as exc: + logger.debug("Serve/dashboard survivor probe failed: %s", exc) + return sorted( + (_without_incarnation(row) for row in planned.values()), + key=lambda row: row["pid"], + ) + survivors = [] + for pid, row in planned.items(): + if pid not in live: + continue + planned_created = row["_create_time"] + live_created = live[pid] + if ( + planned_created is not None + and live_created is not None + and abs(float(live_created) - float(planned_created)) >= 2.0 + ): + # Same number, different process: the pre-update runtime is gone + # and something new registered under its PID. Not a survivor. + continue + survivors.append(_without_incarnation(row)) + return sorted(survivors, key=lambda row: row["pid"]) + + +def _without_incarnation(row: dict) -> dict: + """The operator-facing survivor row (incarnation is a matching key only).""" + return {key: value for key, value in row.items() if key != "_create_time"} + + +def _qualified_serve_skips(skip_units) -> list[dict]: + """Scope-qualify the units the aborted phase already settled. + + ``restarted_scoped_units`` records ``/`` because + ``hermes-serve.service`` can exist in BOTH the user and the system + manager, and those are two different processes. Handing the fresh child a + bare unit name would make one settled scope suppress recovery of the other + — the stale one would never be restarted and nothing downstream would say + so. Entries that carry no scope (a payload built by a pre-update + interpreter) are forwarded without one and read as scope-agnostic there. + """ + rows: list[dict] = [] + for entry in sorted(skip_units or ()): + scope, sep, unit = str(entry).partition("/") + if sep and scope in ("user", "system") and unit: + rows.append({"scope": scope, "unit": unit}) + elif entry: + rows.append({"unit": str(entry)}) + return rows + + +def _recover_gateway_restart_after_abort( + plan, + *, + gateway_mode: bool, + skip_profiles: set[str] | None = None, + skip_units: set[str] | None = None, +) -> dict[str, list]: + """Retry supervised gateway restarts from a clean Python process. + + ``hermes update`` normally performs the fleet restart in the interpreter + that started before ``git pull``. If that phase raises while importing the + new tree, a warning alone leaves the old gateway alive against new files on + disk. The recovery boundary launches the existing per-profile + ``gateway restart`` command through a new interpreter, preserving its + platform-specific drain and service-manager logic without inheriting the + stale ``sys.modules`` graph. + + Only profiles classified as supervisor-owned by the pre-update inventory + are handed off. A manual gateway must remain running and be reported for + explicit operator action rather than being killed without a relaunch + authority; serve/dashboard runtimes from the spawn ledger are likewise + recorded as skipped with a reason instead of vanishing from the pass. + The returned protocol is persisted in the update receipt so operators can + distinguish a spawn failure from a per-profile failure. + + The same child additionally restarts active ``hermes-serve*`` systemd + units (#92145). ``hermes serve`` hosts ``tui_gateway.server`` and is + restarted by the in-process phase alongside the gateway units, but no + per-profile ``gateway restart`` command reaches it — so an abort used to + leave it holding the pre-pull module graph with nothing left to notice. + + ``skip_units`` names the units the aborted phase already settled, as + ``/``. The scope is part of the identity, not decoration: + ``hermes-serve.service`` can exist in both the user and the system + manager as two different processes, and an unqualified token would let a + settled one suppress recovery of a stale one (review on #96235). + + Outcome honesty: ``verified`` means the fresh child independently observed + the profile's systemd unit active after the relaunch. A zero exit from + ``gateway restart`` alone is NOT observed proof that the new code + generation is serving, so those outcomes are reported as + ``relaunch_attempted`` and never claim supervisor coverage. + """ + from hermes_cli.update_cmd import _gateway_recovery_partition + + candidates, skipped = _gateway_recovery_partition( + plan, skip_profiles=skip_profiles + ) + profiles = sorted(candidates) + recover_serve = _serve_unit_recovery_available() + _empty_serve: dict[str, list] = {"verified": [], "failed": []} + if not profiles and not recover_serve: + return { + "requested": [], + "verified": [], + "relaunch_attempted": [], + "failed": [], + "skipped": skipped, + "serve_units": dict(_empty_serve), + } + + def _all_failed() -> dict[str, list]: + return { + "requested": profiles, + "verified": [], + "relaunch_attempted": [], + "failed": profiles, + "skipped": skipped, + "serve_units": dict(_empty_serve), + } + + command = [ + sys.executable, + "-m", + "hermes_cli.update_restart_recovery", + "--stdin", + ] + env = os.environ.copy() + env["HERMES_UPDATE_RESTART_RECOVERY"] = "1" + for marker in ("_HERMES_GATEWAY", "HERMES_GATEWAY", "HERMES_GATEWAY_MODE"): + env.pop(marker, None) + + # A gateway-triggered update may run inside the gateway's systemd cgroup. + # Put the recovery process in a transient user scope before it asks systemd + # to restart that gateway, otherwise KillMode can terminate the recovery + # process together with the old service. If systemd-run is unavailable, + # fail closed rather than pretending the in-cgroup child is independent. + if gateway_mode and sys.platform == "linux": + systemd_run = shutil.which("systemd-run") + if not systemd_run: + logger.warning("Cannot isolate fresh gateway recovery from the gateway cgroup") + return _all_failed() + command = [ + systemd_run, + "--user", + "--scope", + "--quiet", + "--collect", + "--", + *command, + ] + + kwargs = { + "input": json.dumps( + { + "profiles": profiles, + "supervisors": candidates, + "serve_units": { + "recover": recover_serve, + "skip": _qualified_serve_skips(skip_units), + }, + } + ), + "capture_output": True, + "text": True, + "encoding": "utf-8", + "errors": "replace", + "check": False, + "env": env, + # Gateway profiles run sequentially at up to 90s each; the serve pass + # adds its own restart + settle budget on top, so give the child room + # for both rather than killing a recovery that was working. + "timeout": max(180, 30 + 90 * len(profiles) + (150 if recover_serve else 0)), + } + if sys.platform == "win32": + kwargs["creationflags"] = ( + getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + | getattr(subprocess, "DETACHED_PROCESS", 0) + ) + else: + kwargs["start_new_session"] = True + + try: + result = subprocess.run(command, **kwargs) + except (OSError, subprocess.TimeoutExpired) as exc: + logger.warning("Fresh gateway restart recovery failed: %s", exc) + return _all_failed() + + if result.returncode != 0: + logger.warning("Fresh gateway restart recovery exited %s", result.returncode) + return _all_failed() + + try: + recovery_result = json.loads(result.stdout or "") + verified = recovery_result.get("verified") + relaunch_attempted = recovery_result.get("relaunch_attempted") + failed = recovery_result.get("failed") + raw_serve = recovery_result.get("serve_units") or dict(_empty_serve) + except (AttributeError, TypeError, ValueError): + logger.warning("Fresh gateway restart recovery returned invalid JSON") + return _all_failed() + + serve_units = dict(_empty_serve) + if ( + isinstance(raw_serve, dict) + and isinstance(raw_serve.get("verified"), list) + and isinstance(raw_serve.get("failed"), list) + and all( + isinstance(unit, str) + for unit in (*raw_serve["verified"], *raw_serve["failed"]) + ) + ): + serve_units = { + "verified": sorted(raw_serve["verified"]), + "failed": sorted(raw_serve["failed"]), + } + elif recover_serve: + # An unreadable serve block cannot be read as "nothing to do": the + # units it describes are exactly the ones that host tui_gateway and + # may still be serving the pre-update generation. + logger.warning("Fresh recovery returned an invalid serve-unit result") + serve_units = {"verified": [], "failed": [""]} + + buckets = (verified, relaunch_attempted, failed) + reported: list[str] = [] + if all(isinstance(bucket, list) for bucket in buckets): + reported = [*verified, *relaunch_attempted, *failed] + if ( + not all(isinstance(bucket, list) for bucket in buckets) + or any(not isinstance(profile, str) for profile in reported) + or set(reported) != set(profiles) + or len(reported) != len(set(reported)) + ): + logger.warning("Fresh gateway restart recovery returned incomplete profiles") + return _all_failed() + + if verified: + print( + " ✓ Restarted supervised gateway(s) in a fresh process" + " (systemd-verified active): " + ", ".join(sorted(verified)) + ) + if relaunch_attempted: + print( + " ⚠ Relaunch attempted in a fresh process but not" + " supervisor-verified (check these gateways manually): " + + ", ".join(sorted(relaunch_attempted)) + ) + if serve_units["verified"]: + print( + " ✓ Restarted serve unit(s) in a fresh process" + " (new main PID observed): " + + ", ".join(serve_units["verified"]) + ) + if serve_units["failed"]: + print( + " ⚠ Could not verify a replacement for serve unit(s): " + + ", ".join(serve_units["failed"]) + ) + return { + "requested": profiles, + "verified": sorted(verified), + "relaunch_attempted": sorted(relaunch_attempted), + "failed": sorted(failed), + "skipped": skipped, + "serve_units": serve_units, + } + + +def _warn_stale_serve_runtimes(rows) -> None: + """Name the serve/dashboard processes that survived on pre-update code. + + The original #92145 report is a user watching every chat turn fail with an + ``ImportError`` for a symbol that imports fine on disk, with nothing in the + terminal naming the responsible process. ``hermes serve`` hosts + ``tui_gateway.server``; when its unit was never restarted it keeps the + pre-pull ``sys.modules`` graph and there is no gateway row anywhere that + reveals it. Print the PIDs and the exact command that fixes it. + """ + if not rows: + return + print( + " ⚠ These serve/dashboard processes still run pre-update code" + " (they started before the checkout changed):" + ) + for row in rows: + supervisor = row.get("supervisor") or "unknown" + print( + f" pid {row.get('pid')} — {row.get('kind')}" + f" (profile {row.get('profile') or 'default'}, {supervisor})" + ) + print( + " Restart them before using Hermes again, e.g." + " `systemctl --user restart hermes-serve.service`" + " or by relaunching `hermes serve` / the Desktop app." + ) + + +def _abort_recovery_is_complete( + *, + planned_gateway_profiles, + covered_gateway_profiles, + recovery_result, + stale_runtime_rows, +) -> bool: + """May a fresh-process recovery clear the incomplete flag? + + Only when EVERY inventoried runtime family is accounted for. The gateway + leg alone is not enough (#92145): the post-update read-back + (``collect_fleet_versions``) is gateway-only — it reads each profile's + ``gateway_state.json`` / control socket — so a ``hermes serve`` still + holding the pre-update ``sys.modules`` graph is invisible to both the + recovery pass and the verification that follows it. Clearing the flag on + gateway coverage alone is exactly how an update reported success while + every chat turn kept failing with an ``ImportError`` for a symbol that + imports fine on disk. + + Fail closed on each leg: + + * every planned gateway profile is covered, with nothing failed and + nothing merely ``relaunch_attempted`` (rc 0 is not observed coverage); + * no ``hermes-serve*`` unit failed to produce a verified replacement; and + * no serve/dashboard process from the pre-update inventory is still the + same process (``stale_runtime_rows``). + + ``planned_gateway_profiles`` being empty is deliberately NOT completeness: + with no gateway leg to prove, the caller's own fail-closed contract + (``_restart_phase_failure_is_incomplete``) plus the stale-runtime rows + decide the outcome. + """ + if not planned_gateway_profiles: + return False + if not set(planned_gateway_profiles) <= set(covered_gateway_profiles): + return False + result = recovery_result or {} + if result.get("failed") or result.get("relaunch_attempted"): + return False + if (result.get("serve_units") or {}).get("failed"): + return False + return not stale_runtime_rows diff --git a/hermes_cli/update_cmd.py b/hermes_cli/update_cmd.py index 05f47b0d81..0009b70a8a 100644 --- a/hermes_cli/update_cmd.py +++ b/hermes_cli/update_cmd.py @@ -6921,6 +6921,19 @@ def _gateway_service_matches_profile(profile: str, service: object) -> bool: } +# Abort recovery lives in its own bounded module (review on #96235). Re-exported +# here because `hermes_cli.main` and the update flow below address these names +# through `update_cmd`. +from hermes_cli.update_abort_recovery import ( # noqa: E402 + _abort_recovery_is_complete, + _qualified_serve_skips, + _recover_gateway_restart_after_abort, + _serve_unit_recovery_available, + _surviving_pre_update_serve_runtimes, + _warn_stale_serve_runtimes, +) + + def _gateway_recovery_partition( plan, *, skip_profiles: set[str] | None = None ) -> tuple[dict[str, str], list[dict]]: @@ -7020,299 +7033,6 @@ def _gateway_restart_recovery_profiles( return sorted(candidates) -def _serve_unit_recovery_available() -> bool: - """Can a fresh process restart ``hermes-serve*`` units on this host?""" - return sys.platform == "linux" and bool(shutil.which("systemctl")) - - -def _surviving_pre_update_serve_runtimes(plan) -> list[dict]: - """Pre-update serve/dashboard runtimes that are STILL the same process. - - ``ledger_entries()`` re-verifies ``(pid, create_time)`` on every read and - prunes anything that is gone, so an entry that still matches a pre-update - PID is the original process — not a PID-reuse look-alike, and not a - successor (a restarted serve re-registers under a new PID). - - Anything still here after the recovery pass is a live runtime on the - pre-update code generation, which is precisely the unsafe state #92145 - reports. Fail closed: if the ledger cannot be read, every inventoried - runtime counts as surviving, because we cannot prove otherwise. - """ - planned: dict[int, dict] = {} - try: - for runtime in getattr(plan, "runtimes", ()) or (): - if getattr(runtime, "kind", None) not in ("serve", "dashboard"): - continue - pid = getattr(runtime, "pid", None) - if not isinstance(pid, int) or pid <= 0: - continue - planned[pid] = { - "pid": pid, - "kind": str(getattr(runtime, "kind", "")), - "profile": str(getattr(runtime, "profile", "")), - "supervisor": str(getattr(runtime, "supervisor", "")), - } - except Exception as exc: - logger.debug("Could not read planned serve runtimes: %s", exc) - return [] - if not planned: - return [] - try: - from hermes_cli.process_identity import ledger_entries - - live_pids = { - entry.get("pid") - for entry in ledger_entries() - if entry.get("purpose") in ("serve", "dashboard") - } - except Exception as exc: - logger.debug("Serve/dashboard survivor probe failed: %s", exc) - return sorted(planned.values(), key=lambda row: row["pid"]) - return sorted( - (row for pid, row in planned.items() if pid in live_pids), - key=lambda row: row["pid"], - ) - - -def _recover_gateway_restart_after_abort( - plan, - *, - gateway_mode: bool, - skip_profiles: set[str] | None = None, - skip_units: set[str] | None = None, -) -> dict[str, list]: - """Retry supervised gateway restarts from a clean Python process. - - ``hermes update`` normally performs the fleet restart in the interpreter - that started before ``git pull``. If that phase raises while importing the - new tree, a warning alone leaves the old gateway alive against new files on - disk. The recovery boundary launches the existing per-profile - ``gateway restart`` command through a new interpreter, preserving its - platform-specific drain and service-manager logic without inheriting the - stale ``sys.modules`` graph. - - Only profiles classified as supervisor-owned by the pre-update inventory - are handed off. A manual gateway must remain running and be reported for - explicit operator action rather than being killed without a relaunch - authority; serve/dashboard runtimes from the spawn ledger are likewise - recorded as skipped with a reason instead of vanishing from the pass. - The returned protocol is persisted in the update receipt so operators can - distinguish a spawn failure from a per-profile failure. - - The same child additionally restarts active ``hermes-serve*`` systemd - units (#92145). ``hermes serve`` hosts ``tui_gateway.server`` and is - restarted by the in-process phase alongside the gateway units, but no - per-profile ``gateway restart`` command reaches it — so an abort used to - leave it holding the pre-pull module graph with nothing left to notice. - - Outcome honesty: ``verified`` means the fresh child independently observed - the profile's systemd unit active after the relaunch. A zero exit from - ``gateway restart`` alone is NOT observed proof that the new code - generation is serving, so those outcomes are reported as - ``relaunch_attempted`` and never claim supervisor coverage. - """ - candidates, skipped = _gateway_recovery_partition( - plan, skip_profiles=skip_profiles - ) - profiles = sorted(candidates) - recover_serve = _serve_unit_recovery_available() - _empty_serve: dict[str, list] = {"verified": [], "failed": []} - if not profiles and not recover_serve: - return { - "requested": [], - "verified": [], - "relaunch_attempted": [], - "failed": [], - "skipped": skipped, - "serve_units": dict(_empty_serve), - } - - def _all_failed() -> dict[str, list]: - return { - "requested": profiles, - "verified": [], - "relaunch_attempted": [], - "failed": profiles, - "skipped": skipped, - "serve_units": dict(_empty_serve), - } - - command = [ - sys.executable, - "-m", - "hermes_cli.update_restart_recovery", - "--stdin", - ] - env = os.environ.copy() - env["HERMES_UPDATE_RESTART_RECOVERY"] = "1" - for marker in ("_HERMES_GATEWAY", "HERMES_GATEWAY", "HERMES_GATEWAY_MODE"): - env.pop(marker, None) - - # A gateway-triggered update may run inside the gateway's systemd cgroup. - # Put the recovery process in a transient user scope before it asks systemd - # to restart that gateway, otherwise KillMode can terminate the recovery - # process together with the old service. If systemd-run is unavailable, - # fail closed rather than pretending the in-cgroup child is independent. - if gateway_mode and sys.platform == "linux": - systemd_run = shutil.which("systemd-run") - if not systemd_run: - logger.warning("Cannot isolate fresh gateway recovery from the gateway cgroup") - return _all_failed() - command = [ - systemd_run, - "--user", - "--scope", - "--quiet", - "--collect", - "--", - *command, - ] - - kwargs = { - "input": json.dumps( - { - "profiles": profiles, - "supervisors": candidates, - "serve_units": { - "recover": recover_serve, - "skip": sorted(skip_units or set()), - }, - } - ), - "capture_output": True, - "text": True, - "encoding": "utf-8", - "errors": "replace", - "check": False, - "env": env, - # Gateway profiles run sequentially at up to 90s each; the serve pass - # adds its own restart + settle budget on top, so give the child room - # for both rather than killing a recovery that was working. - "timeout": max(180, 30 + 90 * len(profiles) + (150 if recover_serve else 0)), - } - if sys.platform == "win32": - kwargs["creationflags"] = ( - getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) - | getattr(subprocess, "DETACHED_PROCESS", 0) - ) - else: - kwargs["start_new_session"] = True - - try: - result = subprocess.run(command, **kwargs) - except (OSError, subprocess.TimeoutExpired) as exc: - logger.warning("Fresh gateway restart recovery failed: %s", exc) - return _all_failed() - - if result.returncode != 0: - logger.warning("Fresh gateway restart recovery exited %s", result.returncode) - return _all_failed() - - try: - recovery_result = json.loads(result.stdout or "") - verified = recovery_result.get("verified") - relaunch_attempted = recovery_result.get("relaunch_attempted") - failed = recovery_result.get("failed") - raw_serve = recovery_result.get("serve_units") or dict(_empty_serve) - except (AttributeError, TypeError, ValueError): - logger.warning("Fresh gateway restart recovery returned invalid JSON") - return _all_failed() - - serve_units = dict(_empty_serve) - if ( - isinstance(raw_serve, dict) - and isinstance(raw_serve.get("verified"), list) - and isinstance(raw_serve.get("failed"), list) - and all( - isinstance(unit, str) - for unit in (*raw_serve["verified"], *raw_serve["failed"]) - ) - ): - serve_units = { - "verified": sorted(raw_serve["verified"]), - "failed": sorted(raw_serve["failed"]), - } - elif recover_serve: - # An unreadable serve block cannot be read as "nothing to do": the - # units it describes are exactly the ones that host tui_gateway and - # may still be serving the pre-update generation. - logger.warning("Fresh recovery returned an invalid serve-unit result") - serve_units = {"verified": [], "failed": [""]} - - buckets = (verified, relaunch_attempted, failed) - reported: list[str] = [] - if all(isinstance(bucket, list) for bucket in buckets): - reported = [*verified, *relaunch_attempted, *failed] - if ( - not all(isinstance(bucket, list) for bucket in buckets) - or any(not isinstance(profile, str) for profile in reported) - or set(reported) != set(profiles) - or len(reported) != len(set(reported)) - ): - logger.warning("Fresh gateway restart recovery returned incomplete profiles") - return _all_failed() - - if verified: - print( - " ✓ Restarted supervised gateway(s) in a fresh process" - " (systemd-verified active): " + ", ".join(sorted(verified)) - ) - if relaunch_attempted: - print( - " ⚠ Relaunch attempted in a fresh process but not" - " supervisor-verified (check these gateways manually): " - + ", ".join(sorted(relaunch_attempted)) - ) - if serve_units["verified"]: - print( - " ✓ Restarted serve unit(s) in a fresh process" - " (new main PID observed): " - + ", ".join(serve_units["verified"]) - ) - if serve_units["failed"]: - print( - " ⚠ Could not verify a replacement for serve unit(s): " - + ", ".join(serve_units["failed"]) - ) - return { - "requested": profiles, - "verified": sorted(verified), - "relaunch_attempted": sorted(relaunch_attempted), - "failed": sorted(failed), - "skipped": skipped, - "serve_units": serve_units, - } - - -def _warn_stale_serve_runtimes(rows) -> None: - """Name the serve/dashboard processes that survived on pre-update code. - - The original #92145 report is a user watching every chat turn fail with an - ``ImportError`` for a symbol that imports fine on disk, with nothing in the - terminal naming the responsible process. ``hermes serve`` hosts - ``tui_gateway.server``; when its unit was never restarted it keeps the - pre-pull ``sys.modules`` graph and there is no gateway row anywhere that - reveals it. Print the PIDs and the exact command that fixes it. - """ - if not rows: - return - print( - " ⚠ These serve/dashboard processes still run pre-update code" - " (they started before the checkout changed):" - ) - for row in rows: - supervisor = row.get("supervisor") or "unknown" - print( - f" pid {row.get('pid')} — {row.get('kind')}" - f" (profile {row.get('profile') or 'default'}, {supervisor})" - ) - print( - " Restart them before using Hermes again, e.g." - " `systemctl --user restart hermes-serve.service`" - " or by relaunching `hermes serve` / the Desktop app." - ) - - def _warn_gateway_restart_phase_aborted(exc: BaseException, pids) -> None: """Print a recovery warning when the whole restart phase raised. @@ -9718,6 +9438,13 @@ def _cmd_update_impl(args, gateway_mode: bool): # needed to forward already-restarted units to # ``_finish_dashboard_update_cleanup`` (review on #83595). restarted_services: list = [] + # Scope-qualified twin of ``restarted_services`` for systemd units: + # ``user/hermes-serve`` and ``system/hermes-serve`` are two different + # processes, and the abort-recovery child needs to know WHICH one was + # already settled. ``restarted_services`` itself keeps its bare-name + # vocabulary — the fleet probe, the receipt and the operator summary + # all read it (#92145 review). + restarted_scoped_units: set = set() # Keep these restart bookkeeping collections defined even when the # phase raises before its platform-specific imports initialize them. # The abort recovery and the fleet reconciliation both consume the @@ -10256,11 +9983,22 @@ def _cmd_update_impl(args, gateway_mode: bool): f"continuing with remaining gateways" ) - _for_each_systemd_gateway_unit( - result.stdout, - process_unit=_restart_one_systemd_gateway_unit, - on_unit_timeout=_on_unit_timeout, - ) + # Everything this scope appends to ``restarted_services`` + # belongs to THIS systemd manager; qualify it before the + # next scope can append a same-named unit. ``finally`` so a + # phase abort mid-scope still carries the units it settled. + _scope_mark = len(restarted_services) + try: + _for_each_systemd_gateway_unit( + result.stdout, + process_unit=_restart_one_systemd_gateway_unit, + on_unit_timeout=_on_unit_timeout, + ) + finally: + restarted_scoped_units.update( + f"{scope}/{name}" + for name in restarted_services[_scope_mark:] + ) # --- Launchd services (macOS) --- # Restart EVERY ai.hermes.gateway* LaunchAgent, not only the @@ -10519,7 +10257,7 @@ def _cmd_update_impl(args, gateway_mode: bool): _pre_update_plan, gateway_mode=gateway_mode, skip_profiles=_already_restarted_profiles, - skip_units=set(restarted_services), + skip_units=set(restarted_scoped_units), ) _recovery_serve_units = _recovery_result.get("serve_units") or {} _serve_units_failed = list(_recovery_serve_units.get("failed") or []) @@ -10928,50 +10666,6 @@ def _restart_phase_failure_is_incomplete(surviving, pre_restart_pids) -> bool: return pre_restart_pids is None or bool(pre_restart_pids) -def _abort_recovery_is_complete( - *, - planned_gateway_profiles, - covered_gateway_profiles, - recovery_result, - stale_runtime_rows, -) -> bool: - """May a fresh-process recovery clear the incomplete flag? - - Only when EVERY inventoried runtime family is accounted for. The gateway - leg alone is not enough (#92145): the post-update read-back - (``collect_fleet_versions``) is gateway-only — it reads each profile's - ``gateway_state.json`` / control socket — so a ``hermes serve`` still - holding the pre-update ``sys.modules`` graph is invisible to both the - recovery pass and the verification that follows it. Clearing the flag on - gateway coverage alone is exactly how an update reported success while - every chat turn kept failing with an ``ImportError`` for a symbol that - imports fine on disk. - - Fail closed on each leg: - - * every planned gateway profile is covered, with nothing failed and - nothing merely ``relaunch_attempted`` (rc 0 is not observed coverage); - * no ``hermes-serve*`` unit failed to produce a verified replacement; and - * no serve/dashboard process from the pre-update inventory is still the - same process (``stale_runtime_rows``). - - ``planned_gateway_profiles`` being empty is deliberately NOT completeness: - with no gateway leg to prove, the caller's own fail-closed contract - (``_restart_phase_failure_is_incomplete``) plus the stale-runtime rows - decide the outcome. - """ - if not planned_gateway_profiles: - return False - if not set(planned_gateway_profiles) <= set(covered_gateway_profiles): - return False - result = recovery_result or {} - if result.get("failed") or result.get("relaunch_attempted"): - return False - if (result.get("serve_units") or {}).get("failed"): - return False - return not stale_runtime_rows - - def _fleet_probe_expected_runtimes( pre_update_plan, pre_restart_pids, diff --git a/hermes_cli/update_inventory.py b/hermes_cli/update_inventory.py index 9be2dc8108..be03530f7d 100644 --- a/hermes_cli/update_inventory.py +++ b/hermes_cli/update_inventory.py @@ -380,6 +380,11 @@ def collect_runtime_inventory() -> UpdatePlan: "argv": entry.get("argv") or "", "host": entry.get("host") or "", "port": entry.get("port"), + # Process incarnation, not just the numeric PID: a + # post-update survivor probe that compares PIDs alone + # calls a NEW serve that reused the number a survivor + # (#92145 review). + "create_time": entry.get("create_time"), }, ) ) diff --git a/hermes_cli/update_restart_recovery.py b/hermes_cli/update_restart_recovery.py index deed59fd53..ebab536f64 100644 --- a/hermes_cli/update_restart_recovery.py +++ b/hermes_cli/update_restart_recovery.py @@ -33,6 +33,14 @@ covers both and an abort can strand either one (#92145): unit family is enumerated from systemd itself rather than from the update inventory, so a manually launched or Desktop-owned ``hermes serve`` — which has no relaunch authority — can never enter this path. + +Serve-unit identity is always ``/`` (``user/hermes-serve``, +``system/hermes-serve``). The two managers can each own a unit of the same +name and they are different processes: an identity that projects the scope +away lets one settled unit suppress recovery of the other, and lets one +scope's outcome speak for the other's. Scope is therefore never dropped and +reconstructed — not in the skip payload, not in ``verified``/``failed``, not +in the receipt. """ from __future__ import annotations @@ -56,6 +64,7 @@ _PROFILE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") _SUPERVISOR_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,31}$") _UNIT_RE = re.compile(r"^hermes-serve(-[a-z0-9][a-z0-9_-]{0,63})?\.service$") _SERVE_UNIT_PATTERN = "hermes-serve*" +_SCOPE_LABELS = ("user", "system") _UNIT_RESTART_TIMEOUT = 60 _UNIT_SETTLE_ATTEMPTS = 10 _UNIT_SETTLE_DELAY = 1.0 @@ -203,18 +212,23 @@ def restart_profiles( } -def _systemctl_scopes() -> list[list[str]]: +def _systemctl_scopes() -> list[tuple[str, list[str]]]: """``systemctl`` invocations for the user and system scopes, or nothing. Mirrors the scope pair the in-process restart phase walks. ``systemctl`` is resolved through ``shutil.which`` so this module never has to import any Hermes platform helper — importing the freshly pulled tree is exactly what aborted the phase that called us. + + Each scope is returned with its label because ``hermes-serve.service`` in + the user manager and ``hermes-serve.service`` in the system manager are + two different processes. Every identity this module produces or consumes + stays qualified by that label; the bare unit name is never the key. """ systemctl = shutil.which("systemctl") if not systemctl or sys.platform != "linux": return [] - return [[systemctl, "--user"], [systemctl]] + return [("user", [systemctl, "--user"]), ("system", [systemctl])] def _listed_serve_units(scope: list[str], *, run: Callable[..., Any]) -> list[str]: @@ -325,9 +339,48 @@ def _serve_unit_replaced( return False +def _qualified(scope_label: str, base: str) -> str: + """The only identity this module reports for a serve unit.""" + return f"{scope_label}/{base}" + + +def _normalized_skips( + skip_units: Iterable[Any], +) -> tuple[set[tuple[str, str]], set[str]]: + """Split already-settled units into scope-qualified and legacy entries. + + Qualified entries (``{"scope": "user", "unit": "hermes-serve"}`` or the + equivalent ``"user/hermes-serve"`` string) suppress exactly one process. + + A bare ``"hermes-serve"`` carries no scope and therefore cannot say WHICH + of two same-named processes was settled. It is honoured across both scopes + because that is all the information it contains — the caller in this tree + always sends the qualified shape, and this branch exists only for a payload + written by a pre-update interpreter that had no scope to send. + """ + qualified: set[tuple[str, str]] = set() + legacy: set[str] = set() + for entry in skip_units or (): + if isinstance(entry, Mapping): + scope_label = str(entry.get("scope") or "") + base = str(entry.get("unit") or "").removesuffix(".service") + else: + scope_label, sep, unit = str(entry).partition("/") + if not sep: + scope_label, unit = "", scope_label + base = unit.removesuffix(".service") + if not base: + continue + if scope_label in _SCOPE_LABELS: + qualified.add((scope_label, base)) + else: + legacy.add(base) + return qualified, legacy + + def restart_serve_units( *, - skip_units: Iterable[str] = (), + skip_units: Iterable[Any] = (), run: Callable[..., Any] = subprocess.run, sleep: Callable[[float], Any] = time.sleep, ) -> dict[str, list[str]]: @@ -346,24 +399,29 @@ def restart_serve_units( or Desktop-owned ``hermes serve`` owns no unit and therefore cannot be touched here. - Returns ``{"verified": [...], "failed": [...]}`` keyed by base unit name - (no ``.service`` suffix), matching the vocabulary the restart phase - already uses for ``restarted_services``. + Every identity in and out of this function is scope-qualified. The user + manager and the system manager can each own a ``hermes-serve.service``, + and they are different processes: projecting the scope away would let one + already-settled unit suppress recovery of the other, and would let one + scope's success describe the other's outcome. + + Returns ``{"verified": [...], "failed": [...]}`` whose entries are + ``/`` (no ``.service`` suffix), e.g. + ``user/hermes-serve``. """ - skipped = {str(name).removesuffix(".service") for name in skip_units} - # base unit name -> replaced? A unit name can exist in BOTH the user and - # the system scope; each is a separate process and each must be proven. - # Worst outcome wins, so one unprovable scope cannot be masked by the - # other reporting success. - outcomes: dict[str, bool] = {} + skipped_qualified, skipped_legacy = _normalized_skips(skip_units) + # (scope, base unit) -> replaced? A unit name can exist in BOTH the user + # and the system scope; each is a separate process and each is proven, + # reported and accounted for on its own. + outcomes: dict[tuple[str, str], bool] = {} seen: set[tuple[str, str]] = set() - for scope in _systemctl_scopes(): - scope_key = " ".join(scope) + for scope_label, scope in _systemctl_scopes(): for unit in _listed_serve_units(scope, run=run): base = unit.removesuffix(".service") - if (scope_key, base) in seen or base in skipped: + target = (scope_label, base) + if target in seen or target in skipped_qualified or base in skipped_legacy: continue - seen.add((scope_key, base)) + seen.add(target) if not _unit_is_active(scope, unit, run=run): # Not running: nothing is serving a stale generation from it. continue @@ -373,7 +431,7 @@ def restart_serve_units( # be observed, so it cannot be claimed. Restarting blind and # reporting success is the failure mode this module exists to # remove. - outcomes[base] = False + outcomes[target] = False continue try: result = run( @@ -386,21 +444,24 @@ def restart_serve_units( timeout=_UNIT_RESTART_TIMEOUT, ) except (OSError, subprocess.TimeoutExpired): - outcomes[base] = False + outcomes[target] = False continue if getattr(result, "returncode", 1) != 0: # Includes the unprivileged system-scope case. We do not probe # for sudo here: an unverifiable unit must read as failed so # the update stays explicitly incomplete. - outcomes[base] = False + outcomes[target] = False continue - replaced = _serve_unit_replaced( + outcomes[target] = _serve_unit_replaced( scope, unit, previous_pid, run=run, sleep=sleep ) - outcomes[base] = outcomes.get(base, True) and replaced return { - "verified": sorted(base for base, ok in outcomes.items() if ok), - "failed": sorted(base for base, ok in outcomes.items() if not ok), + "verified": sorted( + _qualified(*target) for target, ok in outcomes.items() if ok + ), + "failed": sorted( + _qualified(*target) for target, ok in outcomes.items() if not ok + ), } @@ -435,16 +496,39 @@ def _parse_payload(stream) -> tuple[list[str], dict[str, str], bool, list[str]]: recover_serve = bool(raw_serve.get("recover")) raw_skip = raw_serve.get("skip") or [] if not isinstance(raw_skip, list) or any( - not isinstance(unit, str) for unit in raw_skip + not isinstance(entry, (str, dict)) for entry in raw_skip ): raise ValueError("recovery serve_units skip list is invalid") - # Only the shapes systemd can actually produce for this family; a - # skip entry is a name filter, never a command argument. - skip_units = [ - unit - for unit in raw_skip - if _UNIT_RE.fullmatch(unit if unit.endswith(".service") else f"{unit}.service") - ] + for entry in raw_skip: + # A skip entry names one already-settled process. The qualified + # shape carries the systemd scope, because `hermes-serve.service` + # can exist in BOTH managers and settling one says nothing about + # the other. A bare string is the legacy shape a pre-update + # interpreter can still send; it is kept, and read as + # scope-agnostic by `restart_serve_units`. + if isinstance(entry, dict): + scope_label = entry.get("scope") + unit = entry.get("unit") + if not isinstance(unit, str): + raise ValueError("recovery serve_units skip list is invalid") + if not isinstance(scope_label, str): + scope_label = "" + else: + scope_label, _, unit = entry.rpartition("/") + # Only the shapes systemd can actually produce for this family; a + # skip entry is a name filter, never a command argument. An + # unrecognized scope drops the entry rather than raising: dropping + # a skip can only ever cause one more restart-and-verify, while + # honouring an unreadable one could suppress recovery of a stale + # process. + if scope_label and scope_label not in _SCOPE_LABELS: + continue + if not _UNIT_RE.fullmatch( + unit if unit.endswith(".service") else f"{unit}.service" + ): + continue + base = unit.removesuffix(".service") + skip_units.append(f"{scope_label}/{base}" if scope_label else base) return profiles, supervisors, recover_serve, skip_units diff --git a/tests/hermes_cli/test_serve_runtime_inventory.py b/tests/hermes_cli/test_serve_runtime_inventory.py index ce3ad99cc3..40481462fc 100644 --- a/tests/hermes_cli/test_serve_runtime_inventory.py +++ b/tests/hermes_cli/test_serve_runtime_inventory.py @@ -216,3 +216,22 @@ def test_scan_dashboard_processes_ledger_respects_exclusions(monkeypatch): monkeypatch.setattr(dp.subprocess, "run", lambda *a, **k: fake_run) assert dp._scan_dashboard_processes(exclude_pids={8124}) == [] + + +def test_inventory_records_the_serve_process_incarnation(monkeypatch): + """The plan carries ``(pid, create_time)``, not just the PID (#92145 review). + + The post-abort survivor probe compares a planned serve against the live + spawn ledger. With only the number to compare, a NEW serve that reused the + old PID reads as the pre-update process that never restarted, and recovery + stays incomplete forever. + """ + entry = _ledger_entry(create_time=1712345678.5) + fake_pi = SimpleNamespace( + ledger_entries=lambda **k: [entry], + spawner_is_dead=lambda e: None, + ) + monkeypatch.setitem(sys.modules, "hermes_cli.process_identity", fake_pi) + plan = update_inventory.collect_runtime_inventory() + serves = [r for r in plan.runtimes if r.kind == "serve"] + assert serves and serves[0].detail["create_time"] == 1712345678.5 diff --git a/tests/hermes_cli/test_update_restart_recovery.py b/tests/hermes_cli/test_update_restart_recovery.py index f7bce47ab1..1b3f621803 100644 --- a/tests/hermes_cli/test_update_restart_recovery.py +++ b/tests/hermes_cli/test_update_restart_recovery.py @@ -19,6 +19,7 @@ import sys import textwrap from types import SimpleNamespace +from hermes_cli import update_abort_recovery as abort_recovery from hermes_cli import update_cmd @@ -61,7 +62,7 @@ def test_abort_recovery_hands_managed_profiles_to_a_fresh_process(monkeypatch): calls.append((argv, kwargs)) return _successful_recovery_result(verified=["coder", "default"]) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) plan = SimpleNamespace( runtimes=[ _runtime("default", "systemd"), @@ -136,7 +137,7 @@ def test_abort_recovery_skips_profiles_already_restarted_by_the_phase(monkeypatc calls.append((argv, kwargs)) return _successful_recovery_result(verified=["coder"]) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) plan = SimpleNamespace( runtimes=[_runtime("default", "systemd"), _runtime("coder", "systemd")] ) diff --git a/tests/hermes_cli/test_update_serve_generation_recovery.py b/tests/hermes_cli/test_update_serve_generation_recovery.py index 777e38b94f..609e1ec326 100644 --- a/tests/hermes_cli/test_update_serve_generation_recovery.py +++ b/tests/hermes_cli/test_update_serve_generation_recovery.py @@ -26,6 +26,7 @@ from types import SimpleNamespace import pytest +from hermes_cli import update_abort_recovery as abort_recovery from hermes_cli import update_cmd from hermes_cli import update_restart_recovery as recovery @@ -45,6 +46,18 @@ def _plan(*runtimes): return SimpleNamespace(runtimes=list(runtimes)) +def _serve_runtime(pid, *, create_time=None, kind="serve", profile="default"): + """A planned serve runtime carrying its process incarnation, the way the + inventory records it (``detail["create_time"]`` from the spawn ledger).""" + return SimpleNamespace( + kind=kind, + profile=profile, + supervisor="manual-serve", + pid=pid, + detail={"create_time": create_time}, + ) + + def _identity_module(): return __import__("hermes_cli.process_identity", fromlist=["ledger_entries"]) @@ -331,7 +344,9 @@ def test_serve_unit_verified_when_main_pid_changes(linux_systemctl): main_pids={"hermes-serve.service": 4242}, ) out = recovery.restart_serve_units(run=fake, sleep=lambda _: None) - assert out == {"verified": ["hermes-serve"], "failed": []} + # Identity stays scope-qualified: the same unit name can exist in the + # system manager as a different process (#92145 review). + assert out == {"verified": ["user/hermes-serve"], "failed": []} assert fake.restarted == ["hermes-serve.service"] @@ -351,7 +366,7 @@ def test_unchanged_main_pid_is_not_verified(linux_systemctl): return original_call(argv, **kwargs) out = recovery.restart_serve_units(run=frozen, sleep=lambda _: None) - assert out == {"verified": [], "failed": ["hermes-serve"]} + assert out == {"verified": [], "failed": ["user/hermes-serve"]} def test_unit_that_does_not_come_back_active_is_failed(linux_systemctl): @@ -370,7 +385,7 @@ def test_unit_that_does_not_come_back_active_is_failed(linux_systemctl): return original_call(argv, **kwargs) out = recovery.restart_serve_units(run=dies, sleep=lambda _: None) - assert out == {"verified": [], "failed": ["hermes-serve"]} + assert out == {"verified": [], "failed": ["user/hermes-serve"]} def test_nonzero_restart_is_failed_not_silently_skipped(linux_systemctl): @@ -382,7 +397,7 @@ def test_nonzero_restart_is_failed_not_silently_skipped(linux_systemctl): restart_rc=1, ) out = recovery.restart_serve_units(run=fake, sleep=lambda _: None) - assert out == {"verified": [], "failed": ["hermes-serve"]} + assert out == {"verified": [], "failed": ["user/hermes-serve"]} def test_restart_timeout_is_failed(linux_systemctl): @@ -399,7 +414,7 @@ def test_restart_timeout_is_failed(linux_systemctl): return original_call(argv, **kwargs) out = recovery.restart_serve_units(run=timing_out, sleep=lambda _: None) - assert out == {"verified": [], "failed": ["hermes-serve"]} + assert out == {"verified": [], "failed": ["user/hermes-serve"]} def test_inactive_serve_unit_is_left_alone(linux_systemctl): @@ -421,10 +436,12 @@ def test_units_already_restarted_by_the_phase_are_skipped(linux_systemctl): main_pids={"hermes-serve.service": 1, "hermes-serve-work.service": 2}, ) out = recovery.restart_serve_units( - skip_units=["hermes-serve"], run=fake, sleep=lambda _: None + skip_units=[{"scope": "user", "unit": "hermes-serve"}], + run=fake, + sleep=lambda _: None, ) assert fake.restarted == ["hermes-serve-work.service"] - assert out == {"verified": ["hermes-serve-work"], "failed": []} + assert out == {"verified": ["user/hermes-serve-work"], "failed": []} def test_unrelated_hermes_server_service_is_never_touched(linux_systemctl): @@ -436,7 +453,7 @@ def test_unrelated_hermes_server_service_is_never_touched(linux_systemctl): ) out = recovery.restart_serve_units(run=fake, sleep=lambda _: None) assert fake.restarted == ["hermes-serve.service"] - assert out["verified"] == ["hermes-serve"] + assert out["verified"] == ["user/hermes-serve"] def test_gateway_units_are_not_restarted_by_the_serve_pass(linux_systemctl): @@ -484,7 +501,7 @@ def test_active_unit_with_no_readable_main_pid_is_failed(linux_systemctl): main_pids={"hermes-serve.service": 0}, ) out = recovery.restart_serve_units(run=fake, sleep=lambda _: None) - assert out == {"verified": [], "failed": ["hermes-serve"]} + assert out == {"verified": [], "failed": ["user/hermes-serve"]} assert fake.restarted == [] @@ -492,6 +509,8 @@ def test_same_unit_name_in_both_scopes_is_proven_in_both(linux_systemctl): """User and system scope are two different processes, not one.""" calls: list[tuple[str, list]] = [] + pids = {"user": 10, "system": 20} + def dual_scope(argv, **kwargs): scope = "user" if "--user" in argv else "system" calls.append((scope, list(argv))) @@ -500,17 +519,24 @@ def test_same_unit_name_in_both_scopes_is_proven_in_both(linux_systemctl): if "is-active" in argv: return _Completed(0, stdout="active") if "show" in argv: - return _Completed(0, stdout="10" if scope == "user" else "20") + return _Completed(0, stdout=str(pids[scope])) if "restart" in argv: # The user scope relaunches cleanly; the system scope is refused # (no privilege) and must not be masked by the user-scope success. - return _Completed(0 if scope == "user" else 1) + if scope == "user": + pids["user"] += 1000 + return _Completed(0) + return _Completed(1) raise AssertionError(argv) out = recovery.restart_serve_units(run=dual_scope, sleep=lambda _: None) restarts = [scope for scope, argv in calls if "restart" in argv] assert restarts == ["user", "system"] - assert out == {"verified": [], "failed": ["hermes-serve"]} + # One name, two processes, two independent outcomes. + assert out == { + "verified": ["user/hermes-serve"], + "failed": ["system/hermes-serve"], + } def test_restart_is_invoked_without_an_interactive_auth_prompt(linux_systemctl): @@ -548,6 +574,8 @@ def test_skip_unit_names_are_filtered_to_the_serve_family(): "skip": [ "hermes-serve", "hermes-serve-work.service", + {"scope": "system", "unit": "hermes-serve"}, + {"scope": "user", "unit": "hermes-serve-work.service"}, "hermes-server", "../../etc/systemd/evil", "hermes-serve; rm -rf /", @@ -559,7 +587,13 @@ def test_skip_unit_names_are_filtered_to_the_serve_family(): ) ) assert recover_serve is True - assert skip == ["hermes-serve", "hermes-serve-work.service"] + # Qualified entries keep their scope; the legacy bare shape stays bare. + assert skip == [ + "hermes-serve", + "hermes-serve-work", + "system/hermes-serve", + "user/hermes-serve-work", + ] def test_malformed_serve_block_is_rejected(): @@ -596,7 +630,7 @@ def test_absent_serve_block_defaults_to_no_serve_recovery(): def test_serve_only_fleet_still_spawns_the_recovery_child(monkeypatch): """Before #92145 the driver returned early whenever no gateway profile was supervised, so a stale serve unit was never even attempted.""" - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: True) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: True) calls = [] def fake_run(argv, **kwargs): @@ -613,7 +647,7 @@ def test_serve_only_fleet_still_spawns_the_recovery_child(monkeypatch): ), ) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) result = update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("serve", "default", "manual-serve", 4242)), gateway_mode=False, @@ -627,7 +661,7 @@ def test_serve_only_fleet_still_spawns_the_recovery_child(monkeypatch): def test_child_timeout_budget_covers_the_serve_pass(monkeypatch): """A serve-only recovery must not be killed before its settle window.""" - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: True) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: True) captured = {} def fake_run(argv, **kwargs): @@ -644,7 +678,7 @@ def test_child_timeout_budget_covers_the_serve_pass(monkeypatch): ), ) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("serve", "default", "manual-serve", 4242)), gateway_mode=False, @@ -654,7 +688,7 @@ def test_child_timeout_budget_covers_the_serve_pass(monkeypatch): def test_verified_serve_units_do_not_enter_gateway_restart_vocabulary(monkeypatch): """Serve coverage must not silently widen the gateway fleet-probe gate.""" - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: True) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: True) def fake_run(argv, **kwargs): return _Completed( @@ -669,7 +703,7 @@ def test_verified_serve_units_do_not_enter_gateway_restart_vocabulary(monkeypatc ), ) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) result = update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("gateway", "default", "systemd", 111)), gateway_mode=False, @@ -679,12 +713,12 @@ def test_verified_serve_units_do_not_enter_gateway_restart_vocabulary(monkeypatc def test_no_serve_authority_and_no_gateway_profile_spawns_nothing(monkeypatch): - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: False) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: False) def unreachable(*a, **k): raise AssertionError("recovery child spawned with nothing to recover") - monkeypatch.setattr(update_cmd.subprocess, "run", unreachable) + monkeypatch.setattr(abort_recovery.subprocess, "run", unreachable) result = update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("serve", "default", "manual-serve", 4242)), gateway_mode=False, @@ -694,7 +728,7 @@ def test_no_serve_authority_and_no_gateway_profile_spawns_nothing(monkeypatch): def test_already_restarted_units_are_forwarded_as_skips(monkeypatch): - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: True) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: True) captured = {} def fake_run(argv, **kwargs): @@ -711,18 +745,22 @@ def test_already_restarted_units_are_forwarded_as_skips(monkeypatch): ), ) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("gateway", "default", "systemd", 111)), gateway_mode=False, - skip_units={"hermes-serve"}, + skip_units={"user/hermes-serve"}, ) - assert captured["payload"]["serve_units"]["skip"] == ["hermes-serve"] + # The scope travels with the unit: settling `user/hermes-serve` must not + # suppress recovery of a stale system-scope unit of the same name. + assert captured["payload"]["serve_units"]["skip"] == [ + {"scope": "user", "unit": "hermes-serve"} + ] def test_unreadable_serve_block_from_the_child_reads_as_failure(monkeypatch): """A child that answers with a broken serve block has proven nothing.""" - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: True) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: True) def fake_run(argv, **kwargs): return _Completed( @@ -737,7 +775,7 @@ def test_unreadable_serve_block_from_the_child_reads_as_failure(monkeypatch): ), ) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) result = update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("gateway", "default", "systemd", 111)), gateway_mode=False, @@ -756,7 +794,7 @@ def test_unreadable_serve_block_from_the_child_reads_as_failure(monkeypatch): def test_missing_serve_block_off_linux_is_not_a_failure(monkeypatch): """Hosts with no serve authority must not manufacture a phantom failure.""" - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: False) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: False) def fake_run(argv, **kwargs): return _Completed( @@ -770,7 +808,7 @@ def test_missing_serve_block_off_linux_is_not_a_failure(monkeypatch): ), ) - monkeypatch.setattr(update_cmd.subprocess, "run", fake_run) + monkeypatch.setattr(abort_recovery.subprocess, "run", fake_run) result = update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("gateway", "default", "systemd", 111)), gateway_mode=False, @@ -779,12 +817,12 @@ def test_missing_serve_block_off_linux_is_not_a_failure(monkeypatch): def test_spawn_failure_reports_empty_serve_coverage(monkeypatch): - monkeypatch.setattr(update_cmd, "_serve_unit_recovery_available", lambda: True) + monkeypatch.setattr(abort_recovery, "_serve_unit_recovery_available", lambda: True) def boom(*a, **k): raise OSError("no fork available") - monkeypatch.setattr(update_cmd.subprocess, "run", boom) + monkeypatch.setattr(abort_recovery.subprocess, "run", boom) result = update_cmd._recover_gateway_restart_after_abort( _plan(_runtime("gateway", "default", "systemd", 111)), gateway_mode=False, @@ -814,8 +852,8 @@ def test_serve_coverage_reaches_the_persisted_receipt(tmp_path, monkeypatch): "failed": [], "skipped": [], "serve_units": { - "verified": ["hermes-serve"], - "failed": ["hermes-serve-work"], + "verified": ["user/hermes-serve"], + "failed": ["system/hermes-serve-work"], }, "stale_runtimes": [ { @@ -832,9 +870,11 @@ def test_serve_coverage_reaches_the_persisted_receipt(tmp_path, monkeypatch): persisted = json.loads(path.read_text(encoding="utf-8"))["gateway_restart"][ "fresh_recovery" ] + # The receipt keeps the scope: an operator reading it must know WHICH + # manager owns the unit that could not be proven (#96235 review). assert persisted["serve_units"] == { - "verified": ["hermes-serve"], - "failed": ["hermes-serve-work"], + "verified": ["user/hermes-serve"], + "failed": ["system/hermes-serve-work"], } assert persisted["stale_runtimes"] == [ { @@ -987,3 +1027,201 @@ def test_serve_backend_survives_selection_when_the_dashboard_unit_restarts(monke assert signalled == [7001] assert restarted == ["hermes-serve.service"] assert result["killed"] == [7001] + + +# --------------------------------------------------------------------------- +# Scope-qualified identity (review on #96235) +# --------------------------------------------------------------------------- + + +def _dual_scope_systemctl(*, user_pid=10, system_pid=20, settle=("user", "system")): + """Both managers own a ``hermes-serve.service``; each is its own process.""" + state = {"user": user_pid, "system": system_pid} + calls: list[tuple[str, list]] = [] + + def run(argv, **kwargs): + scope = "user" if "--user" in argv else "system" + calls.append((scope, list(argv))) + if "list-units" in argv: + return _Completed(0, stdout="hermes-serve.service loaded active running x") + if "is-active" in argv: + return _Completed(0, stdout="active") + if "show" in argv: + return _Completed(0, stdout=str(state[scope])) + if "restart" in argv: + if scope in settle: + state[scope] += 1000 + return _Completed(0) + raise AssertionError(argv) + + return run, calls + + +def test_settled_user_scope_does_not_suppress_the_stale_system_scope( + linux_systemctl, +): + """The dual-scope skip collision (review on #96235). + + ``user/hermes-serve`` was restarted before the phase aborted; + ``system/hermes-serve`` is a different process still on the pre-update + generation. A bare ``hermes-serve`` skip token suppressed BOTH, leaving + the stale one running with nothing reporting it. + """ + run, calls = _dual_scope_systemctl() + out = recovery.restart_serve_units( + skip_units=[{"scope": "user", "unit": "hermes-serve"}], + run=run, + sleep=lambda _: None, + ) + restarted_scopes = [scope for scope, argv in calls if "restart" in argv] + assert restarted_scopes == ["system"] + assert out == {"verified": ["system/hermes-serve"], "failed": []} + + +def test_settled_system_scope_does_not_suppress_the_stale_user_scope( + linux_systemctl, +): + """Mirror: authority for one scope is not authority for the other, in + either direction.""" + run, calls = _dual_scope_systemctl() + out = recovery.restart_serve_units( + skip_units=["system/hermes-serve"], + run=run, + sleep=lambda _: None, + ) + restarted_scopes = [scope for scope, argv in calls if "restart" in argv] + assert restarted_scopes == ["user"] + assert out == {"verified": ["user/hermes-serve"], "failed": []} + + +def test_one_authorized_scope_never_mutates_the_same_name_in_the_other( + linux_systemctl, +): + """Name equality is not identity: no systemctl verb at all may reach the + scope that was already settled.""" + run, calls = _dual_scope_systemctl() + recovery.restart_serve_units( + skip_units=[{"scope": "user", "unit": "hermes-serve.service"}], + run=run, + sleep=lambda _: None, + ) + user_verbs = [ + argv for scope, argv in calls if scope == "user" and "list-units" not in argv + ] + assert user_verbs == [], user_verbs + + +def test_each_scope_reports_its_own_outcome(linux_systemctl): + """One scope failing must not be reported as the other's failure.""" + run, _ = _dual_scope_systemctl(settle=("user",)) + out = recovery.restart_serve_units(run=run, sleep=lambda _: None) + assert out == { + "verified": ["user/hermes-serve"], + "failed": ["system/hermes-serve"], + } + + +def test_legacy_unqualified_skip_is_honoured_in_both_scopes(linux_systemctl): + """A payload written by a pre-update interpreter carries no scope. + + That entry is all the information there is, so it suppresses both — the + documented lossy shape, kept only for that skew. The current tree always + sends the qualified form (see + ``test_already_restarted_units_are_forwarded_as_skips``). + """ + run, calls = _dual_scope_systemctl() + out = recovery.restart_serve_units( + skip_units=["hermes-serve"], run=run, sleep=lambda _: None + ) + assert [argv for scope, argv in calls if "restart" in argv] == [] + assert out == {"verified": [], "failed": []} + + +def test_qualified_skip_payload_shape(): + """``restarted_scoped_units`` reaches the child as scope + unit.""" + assert update_cmd._qualified_serve_skips( + {"user/hermes-serve", "system/hermes-serve-work"} + ) == [ + {"scope": "system", "unit": "hermes-serve-work"}, + {"scope": "user", "unit": "hermes-serve"}, + ] + + +def test_unscoped_bookkeeping_entries_are_forwarded_without_a_scope(): + """A name with no scope must not be given one it was never proven for.""" + assert update_cmd._qualified_serve_skips({"hermes-serve"}) == [ + {"unit": "hermes-serve"} + ] + + +def test_scope_qualified_failure_blocks_completion(): + """Completion accounting reads the qualified id, not a bare name.""" + assert ( + update_cmd._abort_recovery_is_complete( + planned_gateway_profiles={"default"}, + covered_gateway_profiles={"default"}, + recovery_result={ + "verified": ["default"], + "relaunch_attempted": [], + "failed": [], + "serve_units": { + "verified": ["user/hermes-serve"], + "failed": ["system/hermes-serve"], + }, + }, + stale_runtime_rows=[], + ) + is False + ) + + +# --------------------------------------------------------------------------- +# Survivor identity is the process incarnation, not the PID number +# --------------------------------------------------------------------------- + + +def test_pid_reuse_by_a_new_serve_is_not_reported_as_the_old_survivor(monkeypatch): + """Same number, later start time: the pre-update process is gone.""" + monkeypatch.setattr( + _identity_module(), + "ledger_entries", + lambda *a, **k: [{"pid": 4242, "purpose": "serve", "create_time": 5000.0}], + ) + assert ( + update_cmd._surviving_pre_update_serve_runtimes( + _plan(_serve_runtime(4242, create_time=1000.0)) + ) + == [] + ) + + +def test_same_pid_and_same_incarnation_is_still_a_survivor(monkeypatch): + monkeypatch.setattr( + _identity_module(), + "ledger_entries", + lambda *a, **k: [{"pid": 4242, "purpose": "serve", "create_time": 1000.5}], + ) + rows = update_cmd._surviving_pre_update_serve_runtimes( + _plan(_serve_runtime(4242, create_time=1000.0)) + ) + assert [row["pid"] for row in rows] == [4242] + assert "_create_time" not in rows[0] + + +def test_missing_incarnation_on_either_side_fails_closed(monkeypatch): + """No incarnation to compare means the runtime cannot be cleared.""" + monkeypatch.setattr( + _identity_module(), + "ledger_entries", + lambda *a, **k: [{"pid": 4242, "purpose": "serve"}], + ) + assert update_cmd._surviving_pre_update_serve_runtimes( + _plan(_serve_runtime(4242, create_time=1000.0)) + ) == [ + { + "pid": 4242, + "kind": "serve", + "profile": "default", + "supervisor": "manual-serve", + } + ]