fix(update): keep manual serve restart obligations visible through gateway recovery

Manual serve/dashboard runtimes with no restart mechanism kept re-arming fleet_restart_pending: every CLI start warned about a completed update. This keeps the false alarm out while never hiding a real obligation: reminders are keyed by (pid, create_time), survive receipt rotation and gateway recovery, and clear only when the incarnation is provably gone or explicitly handed off. The pending marker owns its runtime inventory from the moment it is written; settlement derives coverage only from that inventory, so an older receipt can never discharge a newer update's obligation. Markers without an inventory stay pending. Startup keeps the completed-restart evidence path: a receipt whose restart phase finished vouches for the fleet at its post-update SHA.

Reported-by: adamkrawczyk
Diagnosis credit: KoNit-K (#107237)
Mechanism findings: g3org3yo, kokhlo, fmercurio
Reported-by: andrexibiza (P1 generation ownership)

hermes_cli/update_abort_recovery.py

@@ -227,9 +227,9 @@ def _warn_stale_serve_runtimes(rows) -> None:
         print(
             f"      pid {row.get('pid')} — {row.get('kind')}"
             f" (profile {row.get('profile') or 'default'}, {row.get('supervisor') or 'unknown'})")
-    print(
-        "    Restart them before using Hermes again, e.g. `systemctl --user restart hermes-serve.service`"
-        " or by relaunching `hermes serve` / the Desktop app.")
+    print("    Ask their owner to relaunch `hermes serve` / `hermes dashboard`, or reconnect Desktop for an SSH backend.")
+    if sys.platform == "linux" and any(row.get("supervisor") == "systemd" for row in rows):
+        print("    For unit-managed backends: `systemctl --user restart hermes-serve.service`.")
 
 
 def _owed_stale_serve_rows(rows) -> list[dict]:

hermes_cli/update_cmd.py

@@ -1269,7 +1269,10 @@ def _apply_pulled_update(
     # completed restart leaves this marker so the next update catches up even when git is
     # current. Distinct from ``.update-incomplete`` (venv/install repair).
     # See #95294.
-    _write_fleet_restart_pending_marker(expected_sha=post_pull_sha or "")
+    _write_fleet_restart_pending_marker(
+        expected_sha=post_pull_sha or "",
+        runtimes=_pre_update_plan.to_dict().get("runtimes") if _pre_update_plan is not None else None,
+    )
     # Stale .pyc would ImportError on gateway restart when new source references new names.
     _sweep_bytecode_after_update(branch)
 

hermes_cli/update_cmd_fleet.py

@@ -5,6 +5,7 @@ Split out of ``hermes_cli/update_cmd.py``; every name is re-imported there so
 imported lazily inside each function (no import cycle; test patches stay effective).
 """
 
+import json
 import logging
 from contextlib import suppress
 import os
@@ -50,7 +51,7 @@ def _fleet_restart_pending_marker_path() -> Path:
     return get_hermes_home() / _FLEET_RESTART_PENDING_NAME
 
 
-def _write_fleet_restart_pending_marker(*, expected_sha: str = "") -> None:
+def _write_fleet_restart_pending_marker(*, expected_sha: str = "", runtimes: list[dict] | None = None) -> None:
     """Drop the pull→restart obligation breadcrumb. Never raises."""
     from hermes_cli.update_cmd import _m
     path = _fleet_restart_pending_marker_path()
@@ -61,6 +62,8 @@ def _write_fleet_restart_pending_marker(*, expected_sha: str = "") -> None:
         lines = [f"started={_time.time()}", f"pid={os.getpid()}"]
         if expected_sha:
             lines.append(f"expected_sha={expected_sha}")
+        if runtimes is not None:
+            lines.append("inventory=" + json.dumps({"version": 1, "runtimes": runtimes}))
         path.write_text("\n".join(lines) + "\n", encoding="utf-8")
     except OSError as exc:
         logger.debug("Could not write fleet-restart-pending marker: %s", exc)
@@ -104,7 +107,7 @@ def _receipt_looks_unfinished(receipt: dict) -> bool:
     return bool(receipt.get("stop_reason")) and not succeeded
 
 
-def _receipt_reports_stale_runtime(expected_sha: str | None = None) -> bool:
+def _receipt_reports_stale_runtime(receipt: dict, expected_sha: str | None = None) -> bool:
     """True when ``update_receipts/latest.json`` records a runtime SHA skew.
 
     Prefer the post-restart ``fleet`` matrix. ``plan.runtimes[].code_sha`` is captured
@@ -114,11 +117,6 @@ def _receipt_reports_stale_runtime(expected_sha: str | None = None) -> bool:
     See #95294.
     """
     from hermes_cli.update_cmd import _current_checkout_sha
-    try:
-        from hermes_cli.update_receipt import read_latest_receipt
-        receipt = read_latest_receipt()
-    except Exception:
-        receipt = None
     if not isinstance(receipt, dict):
         return False
     expected_sha = expected_sha or _current_checkout_sha()
@@ -147,50 +145,45 @@ def _receipt_reports_stale_runtime(expected_sha: str | None = None) -> bool:
     )
 
 
-def _receipt_owed_gateways() -> set[tuple[str, str]] | None:
-    """``(kind, profile)`` identities ``latest.json`` owes a current successor.
+def _receipt_owed_gateways(receipt: dict, pending_manual: list[dict]) -> set[tuple[str, str]] | None:
+    """Pure coverage classification after manual retention of this receipt snapshot.
 
-    Empty when the receipt records no runtimes; ``None`` when any recorded runtime is one
-    the gateway matrix cannot vouch for (serve/dashboard, unknown profile).
+    Empty means this receipt owes no gateways, not that an independent marker owes none. Unknown identities and failed manual transfers make coverage unverified.
     """
-    from hermes_cli.update_receipt import read_latest_receipt
-
-    receipt = read_latest_receipt() or {}
     plan = receipt.get("plan") or {}
     entries: list[tuple[object, str | None]] = [(entry, None) for entry in plan.get("runtimes") or []]
+    entries.extend((entry, None) for entry in receipt.get("pending_manual_serves") or [])
     entries.extend((entry, "gateway") for entry in receipt.get("fleet") or [])
     owed: set[tuple[str, str]] = set()
+    unverified = False
     for entry, default_kind in entries:
         if not isinstance(entry, dict):
-            return None
+            unverified = True
+            continue
         kind = entry.get("kind", default_kind)
         profile = entry.get("profile")
+        if kind in ("serve", "dashboard") and entry.get("supervisor") == "manual-serve" and entry not in pending_manual:
+            continue
         if kind != "gateway" or not profile or profile == "unknown":
-            return None
+            unverified = True
+            continue
         owed.add((kind, profile))
-    return owed
+    return None if unverified else owed
 
 
-def _live_fleet_covers_receipt(expected_sha: str | None, *, accept_states: tuple = ("current",)) -> bool:
-    """Require current successors for every recorded runtime, not just any live row.
-
-    A PID changes on restart; the stable identity is (runtime kind, profile).
-    The gateway matrix cannot vouch for serve/dashboard or unidentified runtimes.
-    Keep the historical receipt intact: a manual restart is not a successful update.
-    """
+def _live_fleet_covers_receipt(expected_sha: str | None, receipt: dict, owed: set[tuple[str, str]] | None, *, accept_states: tuple = ("current",)) -> bool:
+    """Require a successor at the expected SHA for every owed gateway identity."""
     if not expected_sha:
         return False
     from hermes_cli.update_receipt import collect_fleet_versions
 
     try:
-        owed = _receipt_owed_gateways()
-        if not owed:
+        if owed is None:
             return False
+        if not owed:
+            return bool((receipt.get("plan") or {}).get("runtimes"))
         fleet = collect_fleet_versions()
-        # ``current``/``stale`` are labels relative to the checkout. A caller asking about the
-        # code a completed update restarted the fleet onto passes ``accept_states`` with
-        # ``stale`` too: the row's stamped ``code_sha`` is the identity that matters there.
-        # ``unknown``/``down`` rows never cover.
+        # State labels are checkout-relative; completed restarts may accept stale rows at the pulled SHA.
         if not fleet or any(
             row.get("state") not in accept_states or row.get("code_sha") != expected_sha
             for row in fleet
@@ -202,55 +195,48 @@ def _live_fleet_covers_receipt(expected_sha: str | None, *, accept_states: tuple
         return False
 
 
-def _read_fleet_marker_expected_sha() -> str:
-    """``expected_sha`` recorded in the pending marker ("" when absent/unreadable)."""
-    with suppress(OSError):
-        for line in _fleet_restart_pending_marker_path().read_text(encoding="utf-8").splitlines():
-            if line.startswith("expected_sha="):
-                return line.split("=", 1)[1].strip()
-    return ""
-
-
 def _marker_only_restart_obsolete() -> bool:
-    """True when the pending marker is a leftover: the fleet already runs the expected code.
+    """Settle only the inventory stored with this marker's target SHA.
 
-    A supervisor-level restart (``systemctl --user restart``, launchctl, ops scripts) never
-    goes through this module's clear path, so the marker survives a restart that DID bring
-    every live gateway to the pulled code — and every later CLI call then prints the
-    interrupted-update warning forever (false positive).
-
-    The marker is not an *unknown* obligation: it records its own ``expected_sha``, so it can
-    be checked against the live fleet directly — no receipt required. Hold it to the same
-    evidence bar ``_live_fleet_covers_receipt`` applies to one: at least one row, and every
-    row a ``current`` gateway under a known profile whose ``code_sha`` equals that
-    ``expected_sha``, with the checkout HEAD not moved past the marker.
-
-    Keep the marker on stale/down rows, on an all-``unknown`` fleet (pre-code-identity
-    gateways cannot prove currency — same conservatism as the silent-failure class
-    #88848/#74973), on a marker with no ``expected_sha``, when a newer pull moved the
-    checkout, and when the probe fails or answers empty.
-
-    A gateway the restart phase stopped and never brought back yields NO row at startup
-    (no ``pre_restart_pids`` → no ``down`` classification), so rows alone cannot prove the
-    whole fleet is back. When ``latest.json`` names the gateways the update owed, every one
-    of them must also be covered by a current row; the rows-only rule applies only when the
-    receipt names none.
+    Historical receipts cannot narrow this obligation. Legacy, malformed or unsupported inventories stay fail-closed; empty discovery never proves a stopped gateway recovered.
     """
-    expected_sha = _read_fleet_marker_expected_sha()
+    try:
+        fields = {}
+        for line in _fleet_restart_pending_marker_path().read_text(encoding="utf-8").splitlines():
+            key, value = line.split("=", 1)
+            if key in fields:
+                return False
+            fields[key] = value
+        expected_sha = fields.get("expected_sha", "").strip()
+        inventory = json.loads(fields.get("inventory", "null"))
+        if not isinstance(inventory, dict) or inventory.get("version") != 1:
+            return False
+        runtimes = inventory.get("runtimes")
+        if not isinstance(runtimes, list) or not runtimes:
+            return False
+        owed = set()
+        for runtime in runtimes:
+            if not isinstance(runtime, dict) or runtime.get("kind") != "gateway":
+                return False
+            profile = runtime.get("profile")
+            if not isinstance(profile, str) or not profile.strip() or profile == "unknown":
+                return False
+            owed.add(("gateway", profile))
+    except (OSError, UnicodeError, ValueError):
+        return False
     if not expected_sha:
-        return False  # pre-expected_sha marker: nothing to verify against
+        return False
     checkout_sha = _current_checkout_sha()
-    if checkout_sha and checkout_sha != expected_sha:
+    if checkout_sha != expected_sha:
         return False  # a newer pull moved HEAD; it owns a fresh obligation
     try:
         from hermes_cli.update_receipt import collect_fleet_versions
         fleet = collect_fleet_versions()
-        owed = _receipt_owed_gateways()
     except Exception as exc:
         logger.debug("Fleet probe failed; keeping fleet-restart-pending marker: %s", exc)
         return False
     if not fleet:
-        return False  # probe answered empty: no proof either way
+        return False  # Absence cannot prove recovery of the recorded inventory.
     for row in fleet:
         if not isinstance(row, dict):
             return False
@@ -259,8 +245,8 @@ def _marker_only_restart_obsolete() -> bool:
             return False  # unidentified runtime: the matrix cannot vouch for it
         if row.get("state") != "current" or str(row.get("code_sha")) != expected_sha:
             return False  # stale / down / unknown-identity row still owes the restart
-    if owed is None or not owed <= {("gateway", row.get("profile")) for row in fleet}:
-        return False  # a gateway the receipt owes is absent (down) or unidentifiable
+    if not owed <= {("gateway", row.get("profile")) for row in fleet}:
+        return False  # A gateway this marker owns is absent.
     _clear_fleet_restart_pending_marker()
     logger.debug(
         "Fleet-restart-pending marker discharged: %d gateway(s) already serve %s",
@@ -270,16 +256,7 @@ def _marker_only_restart_obsolete() -> bool:
 
 
 def _receipt_restart_phase_completed(receipt: dict) -> str | None:
-    """``post_update.sha`` when the receipt's restart phase ran to completion, else None.
-
-    A receipt can be ``failed`` for reasons that have nothing to do with the fleet (a
-    post-restart step crashed, a notice raised) after every gateway was already brought to
-    the pulled code. Its pre-pull ``plan.runtimes[].code_sha`` then no longer describes an
-    obligation; the update owed the fleet ``post_update.sha`` and that is what the live
-    fleet must be checked against. Any later drift (a manual ``git pull`` moving the
-    checkout past a running gateway) belongs to ``gateway/code_skew.py``, not to a warning
-    that blames an update which did restart the fleet.
-    """
+    """Return the pulled SHA when the restart phase completed, even if a later step failed."""
     gateway_restart = receipt.get("gateway_restart")
     if not isinstance(gateway_restart, dict) or not gateway_restart:
         return None
@@ -289,44 +266,47 @@ def _receipt_restart_phase_completed(receipt: dict) -> str | None:
     return str(post_sha) if post_sha else None
 
 
-def _pending_fleet_restart_needed() -> bool:
-    """Reconcile old restart obligations against current, identity-matched gateways.
-
-    Catch-up semantics (``hermes update`` on a current checkout): the fleet must reach the
-    checkout HEAD, whatever moved it there.
-    """
+def _pending_fleet_restart_needed(*, receipt: dict | None = None, pending_manual: list[dict] | None = None) -> bool:
+    """Require identity-matched gateways at checkout HEAD for update catch-up."""
     from hermes_cli.update_cmd import _current_checkout_sha
+    from hermes_cli.update_receipt import read_latest_receipt
+    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
 
-    # The marker has no runtime inventory and may belong to a newer, killed update
-    # than latest.json. An older receipt cannot discharge that unknown obligation.
-    with suppress(OSError):
-        if _fleet_restart_pending_marker_path().is_file():
-            if _marker_only_restart_obsolete():
-                return False
-            return True
-    if not _receipt_reports_stale_runtime():
-        return False
-    return not _live_fleet_covers_receipt(_current_checkout_sha())
-
-
-def _update_owes_fleet_restart() -> bool:
-    """Startup-warning semantics: does the LAST UPDATE still owe the fleet a restart?
-
-    Same evidence as :func:`_pending_fleet_restart_needed`, except that a receipt whose
-    restart phase completed is held to the code it pulled, not to today's checkout: the
-    update kept its promise, and a checkout moved later by hand is not its unfinished work.
-    """
+    if receipt is None:
+        receipt = read_latest_receipt() or {}
+    if pending_manual is None:
+        pending_manual = retain_receipt_manual_serves(receipt)
+    # A marker owns its inventory; latest.json can belong to an older update.
     with suppress(OSError):
         if _fleet_restart_pending_marker_path().is_file():
             return not _marker_only_restart_obsolete()
-    if not _receipt_reports_stale_runtime():
+    owed = _receipt_owed_gateways(receipt, pending_manual)
+    if not _receipt_reports_stale_runtime(receipt):
         return False
+    return not _live_fleet_covers_receipt(_current_checkout_sha(), receipt, owed)
+
+
+def _update_owes_fleet_restart(*, receipt: dict | None = None, pending_manual: list[dict] | None = None) -> bool:
+    """Hold a completed restart to the code it pulled, not a later checkout HEAD."""
     from hermes_cli.update_cmd import _current_checkout_sha
     from hermes_cli.update_receipt import read_latest_receipt
-    restarted_to = _receipt_restart_phase_completed(read_latest_receipt() or {})
+    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
+
+    if receipt is None:
+        receipt = read_latest_receipt() or {}
+    if pending_manual is None:
+        pending_manual = retain_receipt_manual_serves(receipt)
+    # A completed older receipt cannot discharge an independent marker's inventory.
+    with suppress(OSError):
+        if _fleet_restart_pending_marker_path().is_file():
+            return not _marker_only_restart_obsolete()
+    owed = _receipt_owed_gateways(receipt, pending_manual)
+    if not _receipt_reports_stale_runtime(receipt):
+        return False
+    restarted_to = _receipt_restart_phase_completed(receipt)
     if restarted_to:
-        return not _live_fleet_covers_receipt(restarted_to, accept_states=("current", "stale"))
-    return not _live_fleet_covers_receipt(_current_checkout_sha())
+        return not _live_fleet_covers_receipt(restarted_to, receipt, owed, accept_states=("current", "stale"))
+    return not _live_fleet_covers_receipt(_current_checkout_sha(), receipt, owed)
 
 
 def _warn_pending_fleet_restart(*, startup: bool = False) -> None:
@@ -340,9 +320,18 @@ def _warn_pending_fleet_restart(*, startup: bool = False) -> None:
 
 def _warn_pending_fleet_restart_on_startup() -> None:
     """Cheap CLI-startup hint. Never restarts; never raises."""
+    from hermes_cli.update_receipt import read_latest_receipt
+    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves, warn_pending_manual_serves
+
+    receipt = read_latest_receipt() or {}
+    pending_manual = None
     with suppress(Exception):
-        if _update_owes_fleet_restart():
+        pending_manual = retain_receipt_manual_serves(receipt)
+    with suppress(Exception):
+        if _update_owes_fleet_restart(receipt=receipt, pending_manual=pending_manual):
             _warn_pending_fleet_restart(startup=True)
+    with suppress(Exception):
+        warn_pending_manual_serves(startup=True, pending_manual=pending_manual)
 
 
 def _systemd_gateway_unit_listings(on_list_timeout=None):
@@ -1624,6 +1613,12 @@ def _verify_fleet_after_update(restart, *, _pre_update_plan, _windows_gateway_re
                     else None
                 ),
             )
+            from dataclasses import asdict
+            from hermes_cli.update_serve_obligations import defer_manual_serve
+
+            for runtime, outcome in zip(_pre_update_plan.runtimes, _runtime_outcomes):
+                if outcome["outcome"] == "unaccounted" and defer_manual_serve(asdict(runtime), require_alive=True):
+                    outcome["outcome"] = "deferred"
             if report_unaccounted_runtimes(_runtime_outcomes):
                 restart.incomplete = True
             with suppress(Exception):

hermes_cli/update_inventory.py

@@ -400,7 +400,13 @@ def report_unaccounted_runtimes(outcomes: list[dict[str, Any]]) -> bool:
     STALE/DOWN fleet row (exit 1) — a promised restart silently missed is the class this phase
     exists to kill.
     """
-    deferred = [o for o in outcomes if o.get("outcome") == "deferred"]
+    manual = [o for o in outcomes if o.get("outcome") == "deferred" and o.get("mechanism") == "respawn-argv"]
+    if manual:
+        print()
+        print("  ⚠ Manual serve restarts deferred to their owner (reminders retained until the old processes exit):")
+        for o in manual:
+            print(f"    • {o['kind']} [{o['profile']}] pid {o['pid']}: relaunch `hermes serve` / `hermes dashboard`, or reconnect Desktop for an SSH backend")
+    deferred = [o for o in outcomes if o.get("outcome") == "deferred" and o.get("mechanism") != "respawn-argv"]
     if deferred:
         # Surfaced but not escalated: the updater has no authority over these, so holding
         # ``fleet_restart_pending`` for them would never be discharged. See #111494.

hermes_cli/update_receipt.py

@@ -197,6 +197,10 @@ def finalize_update_receipt(outcome: str, fleet: list | None = None, stop_reason
             receipt.data["stop_reason"] = stop_reason
         if fleet is not None:
             receipt.data["fleet"] = fleet
+        from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
+        pending = retain_receipt_manual_serves(read_latest_receipt() or {})
+        if pending:
+            receipt.data["pending_manual_serves"] = pending
         directory = _receipt_dir()
         directory.mkdir(parents=True, exist_ok=True)
         path = directory / f"update_{time.strftime('%Y%m%d_%H%M%S')}_{os.getpid()}.json"

hermes_cli/update_serve_obligations.py

@@ -0,0 +1,92 @@
+"""Durable manual-serve handoffs, independent of gateway restart receipts."""
+
+import json
+import logging
+import math
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+from hermes_constants import get_hermes_home
+
+logger = logging.getLogger(__name__)
+
+
+def defer_manual_serve(runtime: dict, *, require_alive: bool = False) -> bool:
+    """Transfer an identified manual runtime to its own durable restart reminder."""
+    from hermes_cli.process_identity import _pid_alive_matches
+
+    if runtime.get("kind") not in ("serve", "dashboard") or runtime.get("supervisor") != "manual-serve" or runtime.get("restart_via") != "respawn-argv":
+        return False
+    pid = runtime.get("pid")
+    detail = runtime.get("detail")
+    if not isinstance(detail, dict):
+        return False
+    created = detail.get("create_time")
+    if type(pid) is not int or pid <= 0 or type(created) not in (int, float) or not math.isfinite(created) or created <= 0:
+        return False
+    try:
+        alive = _pid_alive_matches(pid, created)
+        if require_alive and alive is not True:
+            return False
+        if alive is False:
+            return True
+        directory = get_hermes_home() / "serve_restart_pending"
+        directory.mkdir(parents=True, exist_ok=True)
+        row = {"kind": runtime["kind"], "profile": runtime.get("profile", "unknown"), "pid": pid, "create_time": created}
+        target = directory / f"{pid}-{float(created).hex()}.json"
+        # One immutable file per incarnation avoids read/merge/write races between CLI startups.
+        temporary = None
+        try:
+            with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=directory, delete=False) as handle:
+                temporary = Path(handle.name)
+                json.dump(row, handle)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temporary, target)
+        finally:
+            if temporary is not None:
+                temporary.unlink(missing_ok=True)
+        return True
+    except (OSError, ValueError, TypeError) as exc:
+        logger.debug("Could not preserve manual serve obligation: %s", exc)
+        return False
+
+
+def retain_receipt_manual_serves(receipt: dict) -> list[dict]:
+    """Return transfers still owed so receipt rotation cannot discard failed writes."""
+    plan = receipt.get("plan") or {}
+    rows = list(plan.get("runtimes") or []) + list(receipt.get("pending_manual_serves") or [])
+    pending = []
+    for row in rows:
+        if not isinstance(row, dict) or row.get("kind") not in ("serve", "dashboard") or row.get("supervisor") != "manual-serve":
+            continue
+        if not defer_manual_serve(row) and row not in pending:
+            pending.append(row)
+    return pending
+
+
+def warn_pending_manual_serves(*, startup: bool = False, pending_manual: list[dict] | None = None) -> None:
+    """Warn about manual debt independently of gateway evidence; optionally reuse a snapshot's failed transfers."""
+    from hermes_cli.process_identity import _pid_alive_matches
+    from hermes_cli.update_receipt import read_latest_receipt
+
+    stream = sys.stderr if startup else sys.stdout
+    if pending_manual is None:
+        pending_manual = retain_receipt_manual_serves(read_latest_receipt() or {})
+    for row in pending_manual:
+        print(f"  ⚠ {row['kind']} [{row.get('profile', 'unknown')}] pid {row.get('pid', 'unknown')}: manual restart reminder could not be saved; restart remains pending in the update receipt.", file=stream)
+        print("    Ask its owner to relaunch `hermes serve` / `hermes dashboard`; check reminder storage permissions and free space.", file=stream)
+    directory = get_hermes_home() / "serve_restart_pending"
+    for path in sorted(directory.glob("*.json")):
+        try:
+            row = json.loads(path.read_text(encoding="utf-8"))
+            if _pid_alive_matches(row["pid"], row["create_time"]) is False:
+                path.unlink(missing_ok=True)
+                continue
+            print(f"  ⚠ {row['kind']} [{row['profile']}] pid {row['pid']}: manual restart still pending; this process may still serve pre-update code.", file=stream)
+            print("    Ask its owner to relaunch `hermes serve` / `hermes dashboard` (reconnect Desktop for an SSH backend).", file=stream)
+        except (OSError, ValueError, KeyError, TypeError) as exc:
+            logger.debug("Could not reconcile manual serve obligation %s: %s", path, exc)
+            print(f"  ⚠ Manual serve restart reminder could not be verified: {path.name}", file=stream)

tests/hermes_cli/test_manual_serve_deferral.py

@@ -0,0 +1,191 @@
+"""Manual serve restart obligations survive gateway completion and receipt rotation (#107224)."""
+
+import json
+from dataclasses import asdict
+
+import pytest
+
+from hermes_cli import process_identity
+from hermes_cli import update_cmd_fleet as fleet
+from hermes_cli import update_receipt
+from hermes_cli.update_inventory import RuntimeRecord, UpdatePlan
+from hermes_constants import get_hermes_home
+
+
+@pytest.mark.parametrize("kind", ["serve", "dashboard"])
+@pytest.mark.parametrize("condition", ["alive", "unknown", "write-error", "missing-identity", "failed-unit"])
+def test_manual_deferral_survives_receipt_rotation(monkeypatch, capsys, kind, condition):
+    runtime = RuntimeRecord(kind=kind, profile="work", pid=900, supervisor="manual-serve", restart_via="respawn-argv", detail={"create_time": 1000.0})
+    plan = UpdatePlan(runtimes=[runtime])
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: True)
+    monkeypatch.setattr(process_identity, "ledger_entries", lambda: [{"pid": 900, "purpose": kind, "create_time": 1000.0}])
+    monkeypatch.setattr(fleet, "_print_legacy_units_warning", lambda: None)
+    monkeypatch.setattr("hermes_cli.update_cmd._finish_dashboard_update_cleanup", lambda *a, **k: None)
+    monkeypatch.setattr("hermes_cli.gateway_migrate.maybe_auto_migrate_after_update", lambda: None)
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: [])
+    restart = fleet._GatewayRestartOutcome(False, [], [], [], [], [], [], set())
+    update_receipt.begin_update_receipt()
+    fleet._write_fleet_restart_pending_marker(expected_sha="new")
+    if condition == "unknown":
+        monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: None)
+    elif condition == "write-error":
+        (get_hermes_home() / "serve_restart_pending").write_text("not a directory")
+    elif condition == "missing-identity":
+        runtime.detail.clear()
+    elif condition == "failed-unit":
+        restart.failed_or_stale_units.append("hermes-serve-work.service")
+        restart.incomplete = True
+    if condition != "alive":
+        with pytest.raises(SystemExit) as exc:
+            fleet._verify_fleet_after_update(restart, _pre_update_plan=plan, _windows_gateway_resume=None, node_failures=[], update_complete=True)
+        assert exc.value.code == 1
+        assert fleet._fleet_restart_pending_marker_path().exists()
+        assert update_receipt.read_latest_receipt()["outcome"] == "partial"
+        return
+    fleet._verify_fleet_after_update(restart, _pre_update_plan=plan, _windows_gateway_resume=None, node_failures=[], update_complete=True)
+    receipt = update_receipt.read_latest_receipt()
+    assert receipt["runtime_outcomes"][0]["outcome"] == "deferred"
+    assert not fleet._fleet_restart_pending_marker_path().exists()
+    assert "hermes-serve.service" not in capsys.readouterr().out
+    update_receipt.begin_update_receipt()
+    update_receipt.finalize_update_receipt("success", fleet=[])
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert f"{kind} [work] pid 900" in warning
+    assert "relaunch" in warning
+    assert "hermes gateway restart" not in warning
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: None)
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "900" in capsys.readouterr().err
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: False)
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "900" not in capsys.readouterr().err
+
+
+@pytest.mark.parametrize("alive", [True, None, False])
+@pytest.mark.parametrize("historical_fleet", ["stale", "current", "empty"])
+@pytest.mark.parametrize("marker", [True, False])
+@pytest.mark.parametrize("gateway_present", [True, False])
+def test_historical_manual_obligation_does_not_block_healthy_gateway(monkeypatch, capsys, alive, historical_fleet, marker, gateway_present):
+    runtime = RuntimeRecord(kind="serve", profile="work", pid=900, supervisor="manual-serve", restart_via="respawn-argv", detail={"create_time": 1000.0})
+    receipt = {"outcome": "partial", "plan": {"runtimes": [asdict(runtime), {"kind": "gateway", "profile": "default"}]}, "fleet": [{"profile": "default", "state": "stale", "code_sha": "old"}]}
+    if historical_fleet == "current":
+        receipt["fleet"] = [{"profile": "default", "state": "current", "code_sha": "new"}]
+    elif historical_fleet == "empty":
+        receipt["fleet"] = []
+    if not gateway_present:
+        receipt["plan"]["runtimes"] = [asdict(runtime)]
+        receipt["fleet"] = []
+    root = get_hermes_home() / "logs" / "update_receipts"
+    root.mkdir(parents=True, exist_ok=True)
+    (root / "latest.json").write_text(json.dumps(receipt))
+    monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: "new")
+    monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: alive)
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: [{"profile": "default", "state": "current", "code_sha": "new"}] if gateway_present else [])
+    if marker:
+        fleet._write_fleet_restart_pending_marker(expected_sha="new")
+    # Legacy markers cannot inherit inventory from any historical receipt, even with a nonempty current fleet.
+    pending = marker
+    assert fleet._pending_fleet_restart_needed() is pending
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert ("serve [work] pid 900" in warning) is (alive is not False)
+    assert ("hermes gateway restart" in warning) is pending
+    assert json.loads((root / "latest.json").read_text()) == receipt
+
+
+@pytest.mark.parametrize("marker", [True, False])
+def test_stamped_manual_only_history_has_no_gateway_obligation(monkeypatch, capsys, marker):
+    runtime = asdict(RuntimeRecord(kind="serve", profile="work", pid=900, code_sha="old", supervisor="manual-serve", restart_via="respawn-argv", detail={"create_time": 1000.0}))
+    receipt = {"outcome": "partial", "plan": {"runtimes": [runtime]}, "fleet": []}
+    root = get_hermes_home() / "logs" / "update_receipts"
+    root.mkdir(parents=True, exist_ok=True)
+    (root / "latest.json").write_text(json.dumps(receipt))
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: True)
+    monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
+    monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: "new")
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: [])
+    if marker:
+        fleet._write_fleet_restart_pending_marker(expected_sha="new")
+    assert fleet._pending_fleet_restart_needed() is marker
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert "serve [work] pid 900" in warning
+    assert ("hermes gateway restart" in warning) is marker
+
+
+@pytest.mark.parametrize("manual_first", [True, False])
+@pytest.mark.parametrize("unsupported", [{"kind": "serve", "supervisor": "desktop"}, {"kind": "gateway", "profile": "unknown"}, None])
+def test_historical_retention_is_independent_of_plan_order(monkeypatch, capsys, manual_first, unsupported):
+    manual = asdict(RuntimeRecord(kind="serve", profile="work", pid=900, supervisor="manual-serve", restart_via="respawn-argv", detail={"create_time": 1000.0}))
+    rows = [manual, unsupported] if manual_first else [unsupported, manual]
+    receipt = {"outcome": "partial", "plan": {"runtimes": [{"kind": "gateway", "profile": "default"}, *rows]}, "fleet": [{"profile": "default", "state": "current", "code_sha": "new"}]}
+    root = get_hermes_home() / "logs" / "update_receipts"
+    root.mkdir(parents=True, exist_ok=True)
+    (root / "latest.json").write_text(json.dumps(receipt))
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: True)
+    monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
+    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
+    pending_manual = retain_receipt_manual_serves(receipt)
+    assert fleet._receipt_owed_gateways(receipt, pending_manual) is None
+    assert list((get_hermes_home() / "serve_restart_pending").glob("*.json"))
+    update_receipt.begin_update_receipt()
+    update_receipt.finalize_update_receipt("success", fleet=[])
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "serve [work] pid 900" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize("failure", ["mkdir", "write", "replace"])
+@pytest.mark.parametrize("gateway_state", ["current", "stale"])
+def test_historical_retention_failure_warns_and_survives_rotation(monkeypatch, capsys, failure, gateway_state):
+    from pathlib import Path
+    from hermes_cli import update_serve_obligations as obligations
+
+    manual = asdict(RuntimeRecord(kind="serve", profile="work", pid=900, supervisor="manual-serve", restart_via="respawn-argv", detail={"create_time": 1000.0}))
+    receipt = {"outcome": "partial", "plan": {"runtimes": [manual]}, "fleet": [{"profile": "default", "state": gateway_state, "code_sha": "new"}]}
+    root = get_hermes_home() / "logs" / "update_receipts"
+    root.mkdir(parents=True, exist_ok=True)
+    (root / "latest.json").write_text(json.dumps(receipt))
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: True)
+    monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: [{"profile": "default", "state": "current", "code_sha": "new"}])
+    directory = get_hermes_home() / "serve_restart_pending"
+
+    def fail(*args, **kwargs):
+        raise OSError("injected persistence failure")
+
+    with monkeypatch.context() as broken:
+        if failure == "mkdir":
+            original = Path.mkdir
+            def mkdir(path, *args, **kwargs):
+                if path == directory:
+                    fail()
+                return original(path, *args, **kwargs)
+            broken.setattr(Path, "mkdir", mkdir)
+        elif failure == "write":
+            broken.setattr(obligations.json, "dump", fail)
+        else:
+            broken.setattr(obligations.os, "replace", fail)
+        fleet._warn_pending_fleet_restart_on_startup()
+        warning = capsys.readouterr().err
+        assert "serve [work] pid 900" in warning
+        assert "could not be saved" in warning
+        assert json.loads((root / "latest.json").read_text()) == receipt
+        for _ in range(2):
+            update_receipt.begin_update_receipt()
+            assert update_receipt.finalize_update_receipt("success", fleet=[]) is not None
+            fleet._warn_pending_fleet_restart_on_startup()
+            assert "serve [work] pid 900" in capsys.readouterr().err
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert "serve [work] pid 900" in warning
+    assert "could not be saved" not in warning
+    assert len(list(directory.iterdir())) == 1
+    update_receipt.begin_update_receipt()
+    update_receipt.finalize_update_receipt("success", fleet=[])
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "serve [work] pid 900" in capsys.readouterr().err
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: False)
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "900" not in capsys.readouterr().err

tests/hermes_cli/test_restart_plan_reconciliation.py

@@ -332,7 +332,7 @@ def test_unaccounted_serve_report_names_serve_remedy_not_gateway_restart(capsys)
     assert report_unaccounted_runtimes(outcomes) is True
     out = capsys.readouterr().out
     assert "serve [default] pid 900" in out
-    assert "hermes-serve.service" in out
+    assert "relaunch `hermes serve`" in out
     assert "hermes gateway restart" not in out
 
 

tests/hermes_cli/test_update_fleet_restart_pending.py

@@ -376,8 +376,8 @@ def test_marker_written_after_pull_cleared_after_successful_restart(
     wrote = []
     orig = update_cmd._write_fleet_restart_pending_marker
 
-    def _spy(*, expected_sha=""):
-        orig(expected_sha=expected_sha)
+    def _spy(*, expected_sha="", runtimes=None):
+        orig(expected_sha=expected_sha, runtimes=runtimes)
         wrote.append(update_cmd._fleet_restart_pending_marker_path().is_file())
 
     monkeypatch.setattr(update_cmd, "_write_fleet_restart_pending_marker", _spy)
@@ -695,7 +695,7 @@ def _patch_marker_sha(monkeypatch, disk_sha):
 
 def test_startup_warn_discharged_when_fleet_current(monkeypatch, capsys):
     disk_sha = "e" * 40
-    update_cmd._write_fleet_restart_pending_marker(expected_sha=disk_sha)
+    update_cmd._write_fleet_restart_pending_marker(expected_sha=disk_sha, runtimes=[{"kind": "gateway", "profile": "default"}])
     _patch_marker_sha(monkeypatch, disk_sha)
     monkeypatch.setattr(
         "hermes_cli.update_receipt.collect_fleet_versions",
@@ -716,13 +716,14 @@ def test_startup_warn_discharged_when_fleet_current(monkeypatch, capsys):
         ("e" * 40, [{"profile": "default", "pid": 42, "code_sha": "7" * 40, "code_version": "0.20.0", "state": "stale"}]),
         ("e" * 40, []),  # probe answered empty: no proof either way
         ("e" * 40, [{"profile": "default", "pid": 42, "code_sha": None, "code_version": None, "state": "unknown"}]),
+        (None, [{"profile": "default", "code_sha": "e" * 40, "state": "current"}]),
         # checkout advanced past the marker: a newer pull owns a fresh obligation
         ("f" * 40, [{"profile": "default", "pid": 42, "code_sha": "e" * 40, "code_version": None, "state": "current"}]),
     ],
-    ids=["stale-row", "empty-probe", "unknown-identity", "checkout-moved"],
+    ids=["stale-row", "empty-probe", "unknown-identity", "unknown-checkout", "checkout-moved"],
 )
 def test_startup_warn_kept_without_positive_evidence(monkeypatch, capsys, disk_sha, fleet):
-    update_cmd._write_fleet_restart_pending_marker(expected_sha="e" * 40)
+    update_cmd._write_fleet_restart_pending_marker(expected_sha="e" * 40, runtimes=[{"kind": "gateway", "profile": "default"}])
     _patch_marker_sha(monkeypatch, disk_sha)
     monkeypatch.setattr("hermes_cli.update_receipt.collect_fleet_versions", lambda **kwargs: fleet)
 
@@ -733,9 +734,9 @@ def test_startup_warn_kept_without_positive_evidence(monkeypatch, capsys, disk_s
 
 
 def test_startup_warn_kept_when_receipt_owed_gateway_is_down(monkeypatch, capsys):
-    """A sibling the restart phase killed yields NO startup row; the receipt still owes it."""
+    """A sibling the restart phase killed yields no startup row; the marker still owns it."""
     disk_sha = "e" * 40
-    update_cmd._write_fleet_restart_pending_marker(expected_sha=disk_sha)
+    update_cmd._write_fleet_restart_pending_marker(expected_sha=disk_sha, runtimes=[{"kind": "gateway", "profile": p} for p in ("alpha", "beta")])
     _patch_marker_sha(monkeypatch, disk_sha)
     receipt_dir = get_hermes_home() / "logs" / "update_receipts"
     receipt_dir.mkdir(parents=True)

tests/hermes_cli/test_update_scoped_reconciliation.py

@@ -0,0 +1,211 @@
+"""Marker settlement uses its own inventory, never a historical receipt's ownership."""
+
+import json
+
+import pytest
+
+from hermes_cli import process_identity, update_cmd_fleet as fleet, update_receipt
+from hermes_constants import get_hermes_home
+
+MANUAL = {"kind": "serve", "profile": "work", "pid": 900, "supervisor": "manual-serve", "restart_via": "respawn-argv", "code_sha": "old", "detail": {"create_time": 1000.0}}
+CURRENT = {"profile": "alpha", "state": "current", "code_sha": "new"}
+GATEWAY = {"kind": "gateway", "profile": "alpha", "code_sha": "old"}
+
+CASES = [
+    ("receipt-successor", {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}, None, [CURRENT], False),
+    ("marker-external-restart", {}, "new", [CURRENT], False),
+    ("missing-sibling", {"outcome": "failed", "plan": {"runtimes": [GATEWAY, dict(GATEWAY, profile="beta")]}}, "new", [CURRENT], True),
+    ("stale-successor", {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}, None, [dict(CURRENT, state="stale", code_sha="old")], True),
+    ("unknown-successor", {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}, None, [dict(CURRENT, state="unknown")], True),
+    ("marker-no-sha", {}, "", [CURRENT], True),
+    ("checkout-moved", {}, "old", [CURRENT], True),
+    ("marker-empty-no-receipt", {}, "new", [], True),
+    ("markerless-stamped-manual", {"outcome": "partial", "plan": {"runtimes": [MANUAL]}}, None, [], False),
+    ("old-manual-new-marker", {"outcome": "success", "post_update": {"sha": "old"}, "plan": {"runtimes": [MANUAL]}, "fleet": []}, "new", [], True),
+    ("same-sha-not-ownership", {"outcome": "success", "post_update": {"sha": "new"}, "plan": {"runtimes": [MANUAL]}, "fleet": []}, "new", [], True),
+    ("mixed-receipt-successors", {"outcome": "partial", "plan": {"runtimes": [GATEWAY, MANUAL]}, "fleet": [dict(CURRENT, state="stale", code_sha="old")]}, None, [CURRENT], False),
+]
+
+
+def seed(monkeypatch, old, marker, live, alive=True):
+    root = get_hermes_home() / "logs" / "update_receipts"
+    root.mkdir(parents=True, exist_ok=True)
+    target = root / "latest.json"
+    target.write_text(json.dumps(old))
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: alive)
+    monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: "new")
+    monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: live)
+    if marker is not None:
+        fleet._write_fleet_restart_pending_marker(expected_sha=marker)
+    return target
+
+
+@pytest.mark.parametrize("name,old,marker,live,pending", CASES, ids=[case[0] for case in CASES])
+def test_scoped_reconciliation_matrix(monkeypatch, capsys, name, old, marker, live, pending):
+    live = list(live)
+    target = seed(monkeypatch, old, marker, live)
+    if name in ("marker-external-restart", "missing-sibling", "checkout-moved"):
+        runtimes = [GATEWAY, dict(GATEWAY, profile="beta")] if name == "missing-sibling" else [GATEWAY]
+        fleet._write_fleet_restart_pending_marker(expected_sha=marker, runtimes=runtimes)
+    before = target.read_bytes()
+    assert fleet._pending_fleet_restart_needed() is pending
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert ("hermes gateway restart" in capsys.readouterr().err) is pending
+    # These unfinished receipts require the same evidence for startup and deferred catch-up.
+    fleet._apply_pending_fleet_restart_catchup(defer=True)
+    assert ("fleet restart deferred" in capsys.readouterr().out) is pending
+    assert target.read_bytes() == before
+    assert fleet._fleet_restart_pending_marker_path().exists() is (marker is not None and pending)
+    if name == "missing-sibling":
+        live.append(dict(CURRENT, profile="beta"))
+        assert not fleet._pending_fleet_restart_needed()
+        assert not fleet._fleet_restart_pending_marker_path().exists()
+        assert target.read_bytes() == before
+
+
+@pytest.mark.parametrize("alive", [True, False, None], ids=["alive", "dead", "unknown"])
+@pytest.mark.parametrize("sha", ["old", "new", None], ids=["different-sha", "same-sha", "no-sha"])
+def test_empty_marker_never_inherits_receipt_ownership(monkeypatch, capsys, alive, sha):
+    old = {"outcome": "success", "plan": {"runtimes": [MANUAL]}, "post_update": {"sha": sha}}
+    target = seed(monkeypatch, old, "new", [], alive)
+    before = target.read_bytes()
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert "hermes gateway restart" in warning
+    assert ("serve [work] pid 900" in warning) is (alive is not False)
+    assert fleet._fleet_restart_pending_marker_path().exists()
+    assert target.read_bytes() == before
+
+
+@pytest.mark.parametrize("consumer", ["predicate", "startup"])
+def test_reconciliation_uses_one_receipt_snapshot(monkeypatch, capsys, consumer):
+    old = {"outcome": "partial", "plan": {"runtimes": [MANUAL]}, "fleet": []}
+    seed(monkeypatch, old, None, [])
+    reads = []
+
+    def read_rotating_receipt():
+        reads.append(True)
+        return old if len(reads) == 1 else {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}
+
+    monkeypatch.setattr(update_receipt, "read_latest_receipt", read_rotating_receipt)
+    if consumer == "predicate":
+        assert not fleet._pending_fleet_restart_needed()
+    else:
+        fleet._warn_pending_fleet_restart_on_startup()
+        warning = capsys.readouterr().err
+        assert "hermes gateway restart" not in warning
+        assert "serve [work] pid 900" in warning
+    assert len(reads) == 1
+    assert list((get_hermes_home() / "serve_restart_pending").glob("*.json"))
+
+
+@pytest.mark.parametrize("marker", [None, "new"])
+@pytest.mark.parametrize("blocked_storage", [False, True])
+def test_probe_exception_does_not_hide_manual_warning(monkeypatch, capsys, marker, blocked_storage):
+    old = {"outcome": "partial", "plan": {"runtimes": [GATEWAY, MANUAL]}}
+    target = seed(monkeypatch, old, marker, [])
+    if marker is not None:
+        fleet._write_fleet_restart_pending_marker(expected_sha=marker, runtimes=[GATEWAY])
+    before = target.read_bytes()
+    if blocked_storage:
+        (get_hermes_home() / "serve_restart_pending").write_text("not a directory")
+
+    def unavailable(**kwargs):
+        raise OSError("gateway probe unavailable")
+
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", unavailable)
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert "hermes gateway restart" in warning
+    assert "serve [work] pid 900" in warning
+    assert ("could not be saved" in warning) is blocked_storage
+    assert target.read_bytes() == before
+
+
+@pytest.mark.parametrize("completed_restart", [False, True])
+@pytest.mark.parametrize("legacy", [False, True], ids=["owned-inventory", "legacy-marker"])
+def test_new_marker_cannot_borrow_old_alpha_receipt(monkeypatch, capsys, legacy, completed_restart):
+    """N owns alpha; N+1 owns alpha and beta but dies before writing its receipt."""
+    old = {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}
+    if completed_restart:
+        old.update(post_update={"sha": "new"}, gateway_restart={"incomplete": False})
+    live = [CURRENT]
+    target = seed(monkeypatch, old, "new", live)
+    marker = fleet._fleet_restart_pending_marker_path()
+    if not legacy:
+        with marker.open("a") as stream:
+            stream.write("inventory=" + json.dumps({"version": 1, "runtimes": [GATEWAY, dict(GATEWAY, profile="beta")]}) + "\n")
+    receipt_before, marker_before = target.read_bytes(), marker.read_bytes()
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "hermes gateway restart" in capsys.readouterr().err
+    assert fleet._pending_fleet_restart_needed()
+    fleet._apply_pending_fleet_restart_catchup(defer=True)
+    assert "fleet restart deferred" in capsys.readouterr().out
+    assert marker.read_bytes() == marker_before
+    assert target.read_bytes() == receipt_before
+    live.append(dict(CURRENT, profile="beta"))
+    assert fleet._pending_fleet_restart_needed() is legacy
+    assert marker.exists() is legacy
+    assert target.read_bytes() == receipt_before
+
+
+@pytest.mark.parametrize("inventory", [None, {}, [], {"version": 2, "runtimes": [GATEWAY]}, {"version": 1, "runtimes": []}, {"version": 1, "runtimes": [GATEWAY, MANUAL]}, {"version": 1, "runtimes": [None]}, {"version": 1, "runtimes": [{"kind": "gateway", "profile": "unknown"}]}, {"version": 1, "runtimes": [{"kind": "gateway", "profile": []}]}])
+def test_unverified_marker_inventory_stays_pending(monkeypatch, inventory):
+    seed(monkeypatch, {"outcome": "success", "plan": {"runtimes": [GATEWAY]}}, "new", [CURRENT])
+    marker = fleet._fleet_restart_pending_marker_path()
+    with marker.open("a") as stream:
+        stream.write("inventory=" + json.dumps(inventory) + "\n")
+    before = marker.read_bytes()
+    assert fleet._pending_fleet_restart_needed()
+    assert marker.read_bytes() == before
+
+
+@pytest.mark.parametrize("suffix", ['inventory={', 'inventory=null\ninventory={"version":1,"runtimes":[]}', 'broken-line'])
+def test_malformed_marker_stays_pending(monkeypatch, suffix):
+    seed(monkeypatch, {}, "new", [CURRENT])
+    marker = fleet._fleet_restart_pending_marker_path()
+    with marker.open("a") as stream:
+        stream.write(suffix + "\n")
+    assert fleet._pending_fleet_restart_needed()
+    assert marker.exists()
+
+
+def test_pulled_update_marker_owns_pre_update_inventory(monkeypatch):
+    from types import SimpleNamespace
+    from hermes_cli import update_cmd
+    from hermes_cli.update_inventory import RuntimeRecord, UpdatePlan
+
+    old = {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}
+    target = seed(monkeypatch, old, None, [CURRENT])
+    before = target.read_bytes()
+    plan = UpdatePlan(runtimes=[RuntimeRecord(kind="gateway", profile=p) for p in ("alpha", "beta")])
+    monkeypatch.setattr(update_cmd, "_invalidate_update_cache", lambda: None)
+    monkeypatch.setattr(update_cmd, "_verify_head_after_pull", lambda *a, **k: "new")
+
+    def interrupt(*args):
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(update_cmd, "_sweep_bytecode_after_update", interrupt)
+    with pytest.raises(KeyboardInterrupt):
+        update_cmd._apply_pulled_update([], "main", "old", SimpleNamespace(in_place_update=True), None, gateway_mode=False, is_fork=False, desktop_dir=None, had_desktop_app_before_update=False, pre_update_snapshot_id=None, _pre_update_plan=plan, _windows_gateway_resume=None, args=SimpleNamespace())
+    marker = fleet._fleet_restart_pending_marker_path()
+    fields = dict(line.split("=", 1) for line in marker.read_text().splitlines())
+    assert fields["expected_sha"] == "new"
+    assert json.loads(fields["inventory"]) == {"version": 1, "runtimes": plan.to_dict()["runtimes"]}
+    assert fleet._pending_fleet_restart_needed()
+    assert target.read_bytes() == before
+
+
+def test_marker_reconciliation_collects_one_live_snapshot(monkeypatch):
+    seed(monkeypatch, {}, "new", [CURRENT])
+    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY])
+    probes = []
+
+    def collect(**kwargs):
+        probes.append(True)
+        return [CURRENT] if len(probes) == 1 else []
+
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", collect)
+    assert not fleet._pending_fleet_restart_needed()
+    assert len(probes) == 1

tests/hermes_cli/test_update_serve_generation_recovery.py

@@ -920,7 +920,7 @@ def test_stale_serve_warning_names_the_process_and_the_fix(capsys):
     out = capsys.readouterr().out
     assert "4242" in out
     assert "serve" in out
-    assert "systemctl --user restart hermes-serve.service" in out
+    assert "systemctl --user restart hermes-serve.service" not in out
 
 
 def test_no_survivors_prints_nothing(capsys):