fix(update): keep serve-unit recovery identity scope-qualified
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YQ9oCBKgAMHSGG8CLEHLMC
This commit is contained in:
@@ -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 ``<scope>/<unit>`` 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
|
||||
``<scope>/<unit>``. 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": ["<unreadable>"]}
|
||||
|
||||
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
|
||||
+37
-343
@@ -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": ["<unreadable>"]}
|
||||
|
||||
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,
|
||||
|
||||
@@ -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"),
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
@@ -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 ``<scope>/<unit>`` (``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
|
||||
``<scope>/<base unit>`` (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
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")]
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user