fix(update): P1 verify owned fleet after catch-up restart

Keep the pending marker until a fresh fleet observation covers its owned gateway inventory. Transfer identified manual serve obligations to durable reminders so an already-verified gateway restart does not keep asking for another restart. Failed transfers retain the marker.

Pin the non-deferred alpha/beta gap and verified-restart-survival cases, including failed probes, storage recovery, and current-successor controls. Both regressions fail without the production change.

Reported-by: ehz0ah

Reported-by: cadamec

hermes_cli/update_cmd_fleet.py

@@ -200,6 +200,8 @@ def _marker_only_restart_obsolete() -> bool:
 
     Historical receipts cannot narrow this obligation. Legacy, malformed or unsupported inventories stay fail-closed; empty discovery never proves a stopped gateway recovered.
     """
+    from hermes_cli.update_serve_obligations import defer_manual_serve
+
     try:
         fields = {}
         for line in _fleet_restart_pending_marker_path().read_text(encoding="utf-8").splitlines():
@@ -216,7 +218,11 @@ def _marker_only_restart_obsolete() -> bool:
             return False
         owed = set()
         for runtime in runtimes:
-            if not isinstance(runtime, dict) or runtime.get("kind") != "gateway":
+            if not isinstance(runtime, dict):
+                return False
+            if runtime.get("kind") in ("serve", "dashboard") and defer_manual_serve(runtime):
+                continue
+            if runtime.get("kind") != "gateway":
                 return False
             profile = runtime.get("profile")
             if not isinstance(profile, str) or not profile.strip() or profile == "unknown":
@@ -519,8 +525,7 @@ def _apply_pending_fleet_restart_catchup(*, defer: bool = False) -> None:
     print()
     _warn_pending_fleet_restart()
     print("→ Running the pending fleet restart...")
-    if _run_pending_fleet_restart():
-        _clear_fleet_restart_pending_marker()
+    if _run_pending_fleet_restart() and not _pending_fleet_restart_needed():
         return
     print("  ⚠ Fleet restart incomplete. Recover with: hermes gateway restart")
     sys.exit(1)

tests/hermes_cli/test_pending_supervisor_recovery.py

@@ -4,11 +4,11 @@ from types import SimpleNamespace
 
 import pytest
 
-from hermes_cli import gateway, main, update_cmd_fleet as fleet
+from hermes_cli import gateway, main, update_cmd_fleet as fleet, update_receipt
 
 
 @pytest.mark.linux_only
-@pytest.mark.parametrize("failure", ["listing", "timeout", "missing", "restart", "inactive", "running", None])
+@pytest.mark.parametrize("failure", ["listing", "timeout", "missing", "restart", "inactive", "running", "missing-owned", None])
 def test_pending_marker_requires_complete_systemd_recovery(monkeypatch, tmp_path, failure):
     stopped = []
     monkeypatch.setattr(gateway, "find_gateway_pids", lambda **kw: [123] if failure == "running" and not stopped else [])
@@ -31,7 +31,7 @@ def test_pending_marker_requires_complete_systemd_recovery(monkeypatch, tmp_path
                 raise FileNotFoundError("systemctl")
             return SimpleNamespace(returncode=int(failure == "listing"), stdout=(
                 "hermes-gateway-one.service loaded active running\n"
-                "hermes-gateway-two.service loaded failed failed\n"), stderr="")
+                + ("" if failure == "missing-owned" else "hermes-gateway-two.service loaded failed failed\n")), stderr="")
         bad = cmd[-1] == "hermes-gateway-two"
         if "restart" in cmd:
             recovered.append(cmd[-1])
@@ -42,8 +42,15 @@ def test_pending_marker_requires_complete_systemd_recovery(monkeypatch, tmp_path
         return SimpleNamespace(returncode=0, stdout="0s")
 
     monkeypatch.setattr(fleet, "_systemctl", systemctl)
+    monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: "pending")
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **kw: [
+        {"profile": name.removeprefix("hermes-gateway-"), "state": "current", "code_sha": "pending"}
+        for name in recovered
+    ])
+    fleet._write_fleet_restart_pending_marker(expected_sha="pending", runtimes=[
+        {"kind": "gateway", "profile": profile} for profile in ("one", "two")
+    ])
     marker = fleet._fleet_restart_pending_marker_path()
-    marker.write_text("expected_sha=pending\n")
     if failure not in (None, "running"):
         with pytest.raises(SystemExit, match="1"):
             fleet._apply_pending_fleet_restart_catchup()

tests/hermes_cli/test_update_fleet_restart_pending.py

@@ -570,9 +570,14 @@ def test_already_up_to_date_runs_pending_restart_when_marker_present(
 ):
     args = _update_args()
     _patch_update_deps(monkeypatch, tmp_path, _make_up_to_date_side_effect())
-    update_cmd._write_fleet_restart_pending_marker(expected_sha="def456")
+    monkeypatch.setattr(update_cmd_fleet, "_current_checkout_sha", lambda: "abc123")
+    update_cmd._write_fleet_restart_pending_marker(expected_sha="abc123", runtimes=[{"kind": "gateway", "profile": "default"}])
 
     seen = {"ran": False}
+    monkeypatch.setattr(
+        "hermes_cli.update_receipt.collect_fleet_versions",
+        lambda **k: [{"profile": "default", "state": "current", "code_sha": "abc123"}] if seen["ran"] else [],
+    )
 
     def _restart():
         seen["ran"] = True
@@ -623,6 +628,10 @@ def test_already_up_to_date_runs_pending_restart_when_receipt_skewed(
     )
 
     seen = {"ran": False}
+    monkeypatch.setattr(
+        "hermes_cli.update_receipt.collect_fleet_versions",
+        lambda **k: [{"profile": "default", "state": "current", "code_sha": disk_sha}] if seen["ran"] else [],
+    )
     monkeypatch.setattr(
         update_cmd,
         "_run_pending_fleet_restart",

tests/hermes_cli/test_update_scoped_reconciliation.py

@@ -150,7 +150,7 @@ def test_new_marker_cannot_borrow_old_alpha_receipt(monkeypatch, capsys, 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": []}]}])
+@pytest.mark.parametrize("inventory", [None, {}, [], {"version": 2, "runtimes": [GATEWAY]}, {"version": 1, "runtimes": []}, {"version": 1, "runtimes": [GATEWAY, dict(MANUAL, detail={})]}, {"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()
@@ -158,6 +158,10 @@ def test_unverified_marker_inventory_stays_pending(monkeypatch, inventory):
         stream.write("inventory=" + json.dumps(inventory) + "\n")
     before = marker.read_bytes()
     assert fleet._pending_fleet_restart_needed()
+    monkeypatch.setattr("hermes_cli.update_cmd._run_pending_fleet_restart", lambda: True)
+    with pytest.raises(SystemExit) as exc:
+        fleet._apply_pending_fleet_restart_catchup()
+    assert exc.value.code == 1
     assert marker.read_bytes() == before
 
 
@@ -197,6 +201,81 @@ def test_pulled_update_marker_owns_pre_update_inventory(monkeypatch):
     assert target.read_bytes() == before
 
 
+@pytest.mark.parametrize("successor", ["missing", "stale", "unknown", "probe-error", "current"])
+def test_catchup_verifies_owned_fleet_after_restart(monkeypatch, successor):
+    from hermes_cli import update_cmd
+
+    old = {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}
+    live = [CURRENT]
+    target = seed(monkeypatch, old, "new", live)
+    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY, dict(GATEWAY, profile="beta")])
+    marker = fleet._fleet_restart_pending_marker_path()
+    receipt_before, marker_before = target.read_bytes(), marker.read_bytes()
+    restarted = []
+
+    def restart():
+        restarted.append("alpha")
+        live[:] = [CURRENT]
+        if successor not in ("missing", "probe-error"):
+            live.append(dict(CURRENT, profile="beta", state=successor, code_sha="old" if successor == "stale" else "new"))
+        return True
+
+    def collect(**kwargs):
+        if restarted and successor == "probe-error":
+            raise OSError("fleet unavailable")
+        return live
+
+    monkeypatch.setattr(update_cmd, "_run_pending_fleet_restart", restart)
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", collect)
+    if successor == "current":
+        fleet._apply_pending_fleet_restart_catchup()
+        assert not marker.exists()
+    else:
+        with pytest.raises(SystemExit) as exc:
+            fleet._apply_pending_fleet_restart_catchup()
+        assert exc.value.code == 1
+        assert marker.read_bytes() == marker_before
+        live[:] = [CURRENT, dict(CURRENT, profile="beta")]
+        monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: live)
+        fleet._apply_pending_fleet_restart_catchup()
+        assert not marker.exists()
+    assert restarted == ["alpha"]
+    assert target.read_bytes() == receipt_before
+
+
+@pytest.mark.parametrize("blocked_storage", [False, True])
+def test_verified_restart_surviving_marker_preserves_manual_debt(monkeypatch, capsys, blocked_storage):
+    from hermes_cli import update_cmd
+
+    manual = [MANUAL, dict(MANUAL, pid=901)]
+    old = {"outcome": "partial", "post_update": {"sha": "new"}, "gateway_restart": {"incomplete": False, "phase_error": ""}, "plan": {"runtimes": [GATEWAY, *manual]}, "fleet": [CURRENT]}
+    target = seed(monkeypatch, old, "new", [CURRENT])
+    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY, *manual])
+    marker = fleet._fleet_restart_pending_marker_path()
+    receipt_before, marker_before = target.read_bytes(), marker.read_bytes()
+    directory = get_hermes_home() / "serve_restart_pending"
+    if blocked_storage:
+        directory.write_text("not a directory")
+    restarted = []
+    monkeypatch.setattr(update_cmd, "_run_pending_fleet_restart", lambda: restarted.append(True) or True)
+    if blocked_storage:
+        with pytest.raises(SystemExit) as exc:
+            fleet._apply_pending_fleet_restart_catchup()
+        assert exc.value.code == 1
+        assert marker.read_bytes() == marker_before
+        directory.unlink()
+        restarted.clear()
+    fleet._apply_pending_fleet_restart_catchup()
+    assert not restarted
+    assert not marker.exists()
+    assert {json.loads(path.read_text())["pid"] for path in directory.glob("*.json")} == {900, 901}
+    fleet._warn_pending_fleet_restart_on_startup()
+    warning = capsys.readouterr().err
+    assert "hermes gateway restart" not in warning
+    assert "pid 900" in warning and "pid 901" in warning
+    assert target.read_bytes() == receipt_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])