fix(update): keep the unsaved manual-serve restart warning while an update runs

The durable running latest.json carries the previous receipt's manual-serve rows
as `carried_manual_serves` (no plan yet), and the startup warning read only
plan.runtimes + pending_manual_serves: the warning vanished during a run and
after a run killed before its plan stage. Read carried rows too, and carry a
killed run's rows forward into the next run.

hermes_cli/update_cmd_fleet.py

@@ -259,6 +259,7 @@ def _receipt_owed_gateways(receipt: dict, pending_manual: list[dict]) -> set[tup
     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, None) for entry in receipt.get("carried_manual_serves") or [])
     entries.extend((entry, "gateway") for entry in receipt.get("fleet") or [])
     owed: set[tuple[str, str]] = set()
     unverified = False

hermes_cli/update_receipt.py

@@ -293,6 +293,7 @@ def begin_update_receipt(*, previous: dict | None = None, correlation_id: str |
                 prior = read_latest_receipt() or {}
                 rows = list((prior.get("plan") or {}).get("runtimes") or [])
                 rows += list(prior.get("pending_manual_serves") or [])
+                rows += list(prior.get("carried_manual_serves") or [])  # a run killed before finalize
                 receipt.data["carried_manual_serves"] = [
                     row for row in rows if isinstance(row, dict) and row.get("supervisor") == "manual-serve"]
     except Exception as exc:  # pragma: no cover - defensive

hermes_cli/update_serve_obligations.py

@@ -60,9 +60,15 @@ def defer_manual_serve(runtime: dict, *, require_alive: bool = False) -> bool:
 
 
 def retain_receipt_manual_serves(receipt: dict) -> list[dict]:
-    """Return transfers still owed so receipt rotation cannot discard failed writes."""
+    """Return transfers still owed so receipt rotation cannot discard failed writes.
+
+    ``carried_manual_serves`` is the durable running record's copy of the previous receipt's rows
+    (update_receipt.begin): read it too, or the warning vanishes while an update runs and after a
+    run killed before its ``plan`` stage.
+    """
     plan = receipt.get("plan") or {}
     rows = list(plan.get("runtimes") or []) + list(receipt.get("pending_manual_serves") or [])
+    rows += list(receipt.get("carried_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":

tests/hermes_cli/test_manual_serve_deferral.py

@@ -209,6 +209,34 @@ def test_historical_retention_failure_warns_and_survives_rotation(monkeypatch, c
     assert "900" not in capsys.readouterr().err
 
 
+def test_unsaved_manual_restart_warning_survives_a_running_and_a_killed_update(monkeypatch, capsys):
+    """The durable running latest.json holds the prior rows as ``carried_manual_serves`` (no plan
+    yet): the startup warning must read them while an update runs, after one is killed before its
+    ``plan`` stage, and the next run must carry them forward again."""
+    manual = asdict(RuntimeRecord(kind="serve", profile="work", pid=900, supervisor="manual-serve", restart_via="respawn-argv", detail={"create_time": 1000.0}))
+    root = get_hermes_home() / "logs" / "update_receipts"
+    root.mkdir(parents=True, exist_ok=True)
+    (root / "latest.json").write_text(json.dumps({"outcome": "partial", "plan": {"runtimes": [manual]}}))
+    (get_hermes_home() / "serve_restart_pending").write_text("not a directory")  # the reminder can't be filed
+    monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: True)
+
+    update_receipt.begin_update_receipt()  # running record replaces latest.json
+    running = json.loads((root / "latest.json").read_text())
+    assert running["outcome"] == "running" and "plan" not in running
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "serve [work] pid 900" in capsys.readouterr().err
+
+    # Killed before the plan stage: the in-memory receipt is gone, latest.json stays the running record.
+    update_receipt._current.set(None)
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert "serve [work] pid 900" in capsys.readouterr().err
+
+    update_receipt.begin_update_receipt()  # the next run reconciles the killed one and carries again
+    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("kind", ["serve", "dashboard"])
 @pytest.mark.parametrize("alive", [True, None, False])
 def test_unreadable_create_time_discharges_only_a_proven_dead_pid(monkeypatch, kind, alive):