fix(update): pause record survives killed recoveries and partner-claimed updates

- A recovering launch killed mid-resume (taskkill, console close) left only its
  .claim file, which nothing read: the paused gateways were never restarted.
  Claims now carry the claimer's identity; a claim whose claimer is dead is
  re-adopted like a dead owner's record. A launch that still owes part of the
  set hands its claim back unowned so the next launch retries at once.
- Under a Desktop/Tauri-driven update the orchestrator holds the marker, so the
  update saw its own partner as "another update", never adopted an orphaned
  pause record and then overwrote it. The update tree holding the checkout lock
  is no longer "elsewhere" (update_lock.update_in_progress for everyone else),
  and write() merges another pause's orphaned set instead of overwriting it
  (refuses when that pause's owner is live).
- Runtime-host detection (gateway run / serve / dashboard in the ancestry)
  uses the canonical command-line matchers, so `--profile X gateway run` and
  the hermes_cli/main.py path form are recognised; the rule sits in
  _adopt_or_refuse's partner check.
- Whole-tree gate: at a moved HEAD only the interrupted-pull marker and current
  dependencies count; a tracked file a build rewrote no longer keeps gateways
  stopped forever. At the pre-update HEAD a new tracked change is still git's
  torn checkout.
- _pause_windows_gateways_for_update split into record / cold-start / stop phases.

(cherry picked from commit a56eaa3a0f65c44f45600a9b2868859d98cc382e)
(cherry picked from commit a72664b05f5f789c0b238ee1fcd1e9e087851182)

hermes_cli/update_cmd_windows.py

@@ -913,31 +913,19 @@ def _pause_windows_gateways_for_update() -> dict | None:
     """Stop running Windows gateways before mutating the checkout or venv.
 
     Scheduled/startup gateways run via pythonw.exe, invisible to the hermes.exe instance guard, yet keep files
-    locked during ``git``/``uv``. Stop only PIDs the gateway discovery code identifies."""
+    locked during ``git``/``uv``. Stop only PIDs the gateway discovery code identifies. The paused set is on
+    disk (``update_pause_record``) before the first stop, with any set a killed earlier update orphaned."""
     from hermes_cli.update_cmd import _m
     if not _m()._is_windows():
         return None
     with _abort_on_error("Could not prepare Windows gateway pause for update"):
-        from gateway.status import get_process_start_time, terminate_pid
+        import gateway.status  # noqa: F401 — fail before the first stop, not half-way through it
         from hermes_cli.gateway import _capture_gateway_argv
     from hermes_cli import update_pause_record as pause_record
-    # A previous update killed mid-pause left its set stopped: bring it back with this run's resume.
-    orphan = pause_record.orphaned_record()
-    adopted = pause_record.drop_never_stopped(dict(orphan["token"])) if orphan else None
+    adopted, claims = pause_record.adopt_orphans()
     profile_processes, service_gateways, service_gateway_pids, running_pids = _discover_windows_gateways()
     if not running_pids:
-        token = _windows_cold_start_plan()
-        # Other profiles may hold a dead attestation even when the active profile owes nothing
-        # (clean exit, autostart not installed): give them a token to ride on.
-        probe = token if token is not None else {"resume_needed": True, "profiles": {}, "unmapped_pids": [], "unmapped": []}
-        _record_attested_cold_start_profiles(probe, set())
-        token = probe if probe.get("cold_start_profiles") else token
-        if adopted is None:
-            return pause_record.stamp_tree(token) if token is not None else None
-        token = pause_record.stamp_tree(pause_record.merge_into(token, adopted))
-        with _abort_on_error("Could not record the adopted paused-gateway set"):
-            pause_record.write(token, owner=pause_record.identity())
-        return token
+        return _cold_start_pause_token(adopted, claims)
     unmapped_pids = [pid for pid in running_pids if pid not in profile_processes and pid not in service_gateway_pids]
     # Snapshot unmapped gateways' argv *before* anything stops so resume (or a later launch) can replay it.
     # Unmapped = no profile->PID-file mapping (e.g. Scheduled Task ``pythonw.exe -m ...``).
@@ -946,7 +934,6 @@ def _pause_windows_gateways_for_update() -> dict | None:
                                             "Could not capture argv for unmapped gateway %s: %s", int(pid))}
         for pid in unmapped_pids
     ]
-    # Durable BEFORE the first stop: a killed updater must leave a record the next launch can resume.
     intended = {
         "resume_needed": True, "unmapped_pids": unmapped_pids, "unmapped": unmapped,
         "profiles": {str(profile_processes[pid].profile): int(pid) for pid in running_pids
@@ -957,11 +944,43 @@ def _pause_windows_gateways_for_update() -> dict | None:
         names = [str(s.name) for s in service_gateways]
         intended.update(services=names, expected_services=list(names), restarted_services=[],
                         service_profiles={str(s.name): str(s.profile) for s in service_gateways})
-    if adopted is not None:
-        pause_record.merge_into(intended, adopted)
-    pause_record.stamp_tree(intended)
     with _abort_on_error("Could not record the gateways this update pauses"):
-        pause_record.write(intended, owner=pause_record.identity())
+        pause_record.record_pause(intended, adopted, claims)
+    profiles = _stop_windows_gateways(running_pids, profile_processes, service_gateway_pids, unmapped_pids, unmapped)
+    token = {"resume_needed": True, "profiles": profiles, "unmapped_pids": unmapped_pids, "unmapped": unmapped}
+    # Every profile with ANY live gateway at discovery counts as running: service-supervised ones skip the
+    # socket pause (absent from ``profiles``) but the SCM restart brings them back, not a cold-start.
+    running_profiles = set(profiles) | {str(p.profile) for p in profile_processes.values()} | {str(s.profile) for s in service_gateways}
+    _record_attested_cold_start_profiles(token, running_profiles)
+    try:
+        token = _pause_windows_gateway_services(service_gateways, token, profiles, unmapped)
+    except RuntimeError as exc:
+        if "rollback failures" not in str(exc):  # everything stopped was restarted in-line
+            pause_record.discharge(intended)
+        raise
+    return pause_record.finish_pause(token, intended, adopted)
+
+
+def _cold_start_pause_token(adopted: dict | None, claims: list) -> dict | None:
+    """No gateway runs: the cold-start plan, plus any orphaned set adopted from a killed update."""
+    from hermes_cli import update_pause_record as pause_record
+    token = _windows_cold_start_plan()
+    # Other profiles may hold a dead attestation even when the active profile owes nothing
+    # (clean exit, autostart not installed): give them a token to ride on.
+    probe = token if token is not None else {"resume_needed": True, "profiles": {}, "unmapped_pids": [], "unmapped": []}
+    _record_attested_cold_start_profiles(probe, set())
+    token = probe if probe.get("cold_start_profiles") else token
+    if adopted is None:
+        return pause_record.stamp_tree(token) if token is not None else None
+    token = token if token is not None else {"resume_needed": True, "profiles": {}, "unmapped_pids": [], "unmapped": []}
+    with _abort_on_error("Could not record the adopted paused-gateway set"):
+        return pause_record.record_pause(token, adopted, claims)
+
+
+def _stop_windows_gateways(running_pids, profile_processes, service_gateway_pids, unmapped_pids, unmapped) -> dict:
+    """Socket-pause the mapped gateways, drain, then tree-kill survivors; the paused ``profiles``."""
+    from gateway.status import get_process_start_time, terminate_pid
+    from hermes_cli.update_cmd import _m
     profiles, mapped_pids, socket_acks = _request_socket_pauses(running_pids, profile_processes, service_gateway_pids)
     # Resolve venv-side launchers BEFORE draining: a dead worker's parent cannot be recovered (NoSuchProcess).
     # The launcher keeps ``.pyd`` mapped and would trip the venv-holder guard; it is killed with the survivors.
@@ -984,22 +1003,7 @@ def _pause_windows_gateways_for_update() -> dict | None:
         print(f"  → Stopped {len(unmapped_pids)} gateway process(es) without profile mapping")
         if any(not u.get("argv") for u in unmapped):  # no recoverable cmdline (psutil missing, denied, gone)
             print("    Restart manually after update: hermes gateway run")
-    token = {"resume_needed": True, "profiles": profiles, "unmapped_pids": unmapped_pids, "unmapped": unmapped}
-    # Every profile with ANY live gateway at discovery counts as running: service-supervised ones skip the
-    # socket pause (absent from ``profiles``) but the SCM restart brings them back, not a cold-start.
-    running_profiles = set(profiles) | {str(p.profile) for p in profile_processes.values()} | {str(s.profile) for s in service_gateways}
-    _record_attested_cold_start_profiles(token, running_profiles)
-    try:
-        token = _pause_windows_gateway_services(service_gateways, token, profiles, unmapped)
-    except RuntimeError as exc:
-        if "rollback failures" not in str(exc):  # everything stopped was restarted in-line
-            pause_record.discharge(intended)
-        raise
-    if adopted is not None:
-        pause_record.merge_into(token, adopted)
-    token.update({key: intended[key] for key in ("pause_id", "pre_sha", "dirty_at_pause", "identities") if key in intended})
-    pause_record.sync(token)
-    return token
+    return profiles
 
 
 def _record_attested_cold_start_profiles(token: dict, running_profiles: set) -> None:

hermes_cli/update_lock.py

@@ -1305,6 +1305,11 @@ def update_tree_job() -> int:
     return job
 
 
+def holds_checkout_lock(install_root: Path | str | None = None) -> bool:
+    """True when this process holds (or joined) the checkout lock: it IS the running update."""
+    return _HELD is not None and os.path.realpath(_HELD["path"]) == os.path.realpath(checkout_lock_path(install_root))
+
+
 def update_in_progress(install_root: Path | str | None = None) -> bool:
     """True while an update owns this install: a LIVE marker or a held checkout lock."""
     return read_live_update(install_root=install_root) is not None or checkout_lock_held(install_root)
@@ -1322,9 +1327,9 @@ def checkout_lock_held(install_root: Path | str | None = None) -> bool:
     launch the others would park. A lock call that fails outright (ENOLCK on NFS without lockd,
     EOPNOTSUPP on some SMB shares) means nothing can hold it: free, so the interrupted-pull repair
     runs unguarded there as designed."""
-    path = checkout_lock_path(install_root)
-    if _HELD is not None and _HELD["path"] == str(path):
+    if holds_checkout_lock(install_root):
         return True
+    path = checkout_lock_path(install_root)
     try:
         fd = os.open(path, os.O_RDONLY | getattr(os, "O_BINARY", 0))
     except FileNotFoundError:

hermes_cli/update_pause_record.py

@@ -7,14 +7,19 @@ home BEFORE anything is stopped, names its owner (pid + creation time, the updat
 identity format), is rewritten as the token changes, and is removed only after a verified
 resume. A later launch whose owner check finds the updater dead (and no update live) resumes it.
 
-Resuming is gated on a whole tree: no interrupted-pull marker, no tracked change beyond the ones
-present at pause time, and HEAD still at the pre-update commit or dependencies current for the
-moved HEAD. Otherwise the record stays for the next launch, which runs after the interrupted-pull
-restore and the dependency sync.
+Resuming is gated on a whole tree: no interrupted-pull marker; at the pre-update HEAD, no tracked
+change beyond the ones present at pause time (git died before moving HEAD); at a moved HEAD,
+dependencies current for it (a build step may rewrite tracked files there). Otherwise the record
+stays for the next launch, which runs after the interrupted-pull restore and the dependency sync.
+
+A recovering launch claims the record by renaming it to ``<record>.<pid>.<nonce>.claim`` and names
+itself inside; a claim whose claimer died (killed mid-resume) is re-adopted like a dead owner's
+record. ``write`` never overwrites another pause: an orphan's set is merged in, a live one refuses.
 """
 
 from __future__ import annotations
 
+import glob
 import json
 import os
 import subprocess
@@ -130,16 +135,17 @@ def tree_is_whole(token: dict, root: Path | None = None) -> tuple[bool, str]:
     if (root / ".git").exists():
         if interrupted_pull_marker(root).exists():
             return False, "the checkout is mid-pull (interrupted-pull marker present)"
-        changes = tracked_changes(root)
-        if changes is None:
-            return False, "git cannot read the checkout state"
-        unexpected = sorted(set(changes) - set(token.get("dirty_at_pause") or []))
-        if unexpected:
-            return False, f"the checkout has {len(unexpected)} file(s) git left half-written (e.g. {unexpected[0]})"
         head = head_sha(root)
         if head is None:
             return False, "the checkout HEAD is unreadable"
         if head == token.get("pre_sha"):
+            # HEAD never moved: a new tracked change is git's half-written checkout.
+            changes = tracked_changes(root)
+            if changes is None:
+                return False, "git cannot read the checkout state"
+            unexpected = sorted(set(changes) - set(token.get("dirty_at_pause") or []))
+            if unexpected:
+                return False, f"the checkout has {len(unexpected)} file(s) git left half-written (e.g. {unexpected[0]})"
             return True, ""
     if not _venv_is_current(root):
         return False, "dependencies are not current for the updated code yet"
@@ -165,15 +171,25 @@ def read(path: Path | None = None) -> dict | None:
     return body if isinstance(body, dict) and isinstance(body.get("token"), dict) else None
 
 
+UNOWNED = {"pid": 0, "ct": None}
+
+
+class RecordConflict(OSError):
+    """The record names another pause whose owner is still live."""
+
+
 def write(token: dict, *, owner: dict | None = None, path: Path | None = None) -> None:
-    """Persist *token*; the owner of an existing record for the same pause is kept."""
+    """Persist *token*. The owner of the same pause is kept; another pause's orphaned set is merged
+    into *token* (in place, so this run's resume brings it back); a live one refuses."""
     path = path or record_path()
     existing = read(path)
+    same = existing is not None and existing["token"].get("pause_id") == token.get("pause_id")
+    if existing is not None and not same:
+        if identity_is_live(existing.get("owner")):
+            raise RecordConflict(f"{path} holds gateways paused by live process {existing['owner'].get('pid')}")
+        merge_into(token, drop_never_stopped(dict(existing["token"])))
     if owner is None:
-        if existing is not None and existing["token"].get("pause_id") == token.get("pause_id"):
-            owner = existing["owner"]
-        else:
-            owner = identity()
+        owner = existing["owner"] if same else identity()
     _atomic_write(path, {"schema": 1, "owner": owner, "install_root": str(install_root()), "token": token})
 
 
@@ -188,8 +204,8 @@ def discharge(token: dict, path: Path | None = None) -> None:
 
 def sync(token: dict) -> None:
     """Mirror the token after a resume attempt: done → delete; anything still owed → rewrite."""
-    if not token.get("pause_id"):
-        return
+    if not token.get("pause_id") or token.get("recovery"):
+        return  # a recovering launch keeps what it owes in its own claim (_resume_claimed)
     try:
         if token.get("resume_needed") or token.get("resume_deferred"):
             owed = {**token, "resume_needed": True}
@@ -202,21 +218,103 @@ def sync(token: dict) -> None:
 
 
 def _live_update_elsewhere() -> bool:
-    # TODO(LP-LOCK): switch to update_lock.update_in_progress(install_root) once it lands; the
-    # marker read below is the existing path and misses an orphaned completion child.
-    from hermes_cli.update_lock import read_live_update
-    holder = read_live_update()
-    return holder is not None and holder.pid != os.getpid()
+    """A live update other than the one this process belongs to. The update tree holding the
+    checkout lock (this process plus the partner whose marker it adopted) is not "elsewhere"."""
+    from hermes_cli import update_lock
+    root = install_root()
+    return not update_lock.holds_checkout_lock(root) and update_lock.update_in_progress(root)
+
+
+def _claims(path: Path) -> list[Path]:
+    return sorted(path.parent.glob(f"{glob.escape(path.name)}.*.claim"))
+
+
+def _claimer(claim: Path, body: dict) -> dict:
+    """Who holds a claim: the identity it wrote, else (killed before writing it) its name's pid."""
+    if isinstance(body.get("claimer"), dict):
+        return body["claimer"]
+    pid = claim.name[len(RECORD_NAME) + 1:].split(".", 1)[0]
+    return {"pid": int(pid) if pid.isdigit() else 0, "ct": None}
+
+
+def orphans(path: Path | None = None) -> list[tuple[Path, dict]]:
+    """``(file, body)`` for this install's record whose owner is dead and every claim whose claimer
+    is dead — empty while another update is live (it adopts them itself)."""
+    path = path or record_path()
+    found = []
+    body = read(path)
+    if body is not None and not identity_is_live(body.get("owner")):
+        found.append((path, body))
+    for claim in _claims(path):
+        claimed = read(claim)
+        if claimed is not None and not identity_is_live(_claimer(claim, claimed)):
+            found.append((claim, claimed))
+    found = [(src, b) for src, b in found if b.get("install_root") == str(install_root())]
+    if not found or _live_update_elsewhere():
+        return []
+    return found
 
 
 def orphaned_record(path: Path | None = None) -> dict | None:
-    """The record when its owner is dead, it belongs to this install and no other update is live."""
-    body = read(path)
-    if body is None or body.get("install_root") != str(install_root()):
+    found = orphans(path)
+    return found[0][1] if found else None
+
+
+def claim(src: Path) -> tuple[Path, dict] | None:
+    """Take *src* (an orphaned record or a dead claimer's claim) for this process; ``None`` when a
+    concurrent launch won it. The rename is the arbiter; the claimer line names our incarnation."""
+    path = record_path()
+    mine = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}.claim")
+    try:
+        os.rename(src, mine)
+    except OSError:
         return None
-    if identity_is_live(body.get("owner")) or _live_update_elsewhere():
+    body = read(mine)
+    if body is None:
+        with suppress(OSError):
+            mine.unlink()
         return None
-    return body
+    body["claimer"] = identity()
+    _atomic_write(mine, body)
+    return mine, body
+
+
+def adopt_orphans() -> tuple[dict | None, list[Path]]:
+    """For ``hermes update``: claim every orphaned pause and merge their sets. The caller records the
+    merged set durably, then :func:`release_claims` — a crash in between leaves claims to re-adopt."""
+    adopted, claims = None, []
+    for src, _body in orphans():
+        won = claim(src)
+        if won is None:
+            continue
+        claims.append(won[0])
+        adopted = merge_into(adopted, drop_never_stopped(dict(won[1]["token"])))
+    return adopted, claims
+
+
+def release_claims(claims: list[Path]) -> None:
+    for path in claims:
+        with suppress(OSError):
+            path.unlink()
+
+
+def record_pause(token: dict, adopted: dict | None, claims: list[Path]) -> dict:
+    """Merge the adopted set, stamp the tree and persist under this process BEFORE the first stop."""
+    if adopted is not None:
+        merge_into(token, adopted)
+    stamp_tree(token)
+    write(token, owner=identity())
+    release_claims(claims)
+    return token
+
+
+def finish_pause(token: dict, intended: dict, adopted: dict | None) -> dict:
+    """Carry the recorded pause identity onto the final token and mirror it to disk."""
+    if adopted is not None:
+        merge_into(token, adopted)
+    token.update({key: intended[key] for key in ("pause_id", "pre_sha", "dirty_at_pause", "identities") if key in intended})
+    sync(token)
+    return token
 
 
 def drop_never_stopped(token: dict) -> dict:
@@ -235,7 +333,8 @@ def merge_into(token: dict | None, adopted: dict) -> dict:
     profiles = token.setdefault("profiles", {})
     for name, pid in (adopted.get("profiles") or {}).items():
         profiles.setdefault(name, pid)
-    token.setdefault("unmapped", []).extend(adopted.get("unmapped") or [])
+    unmapped = token.setdefault("unmapped", [])
+    unmapped.extend(u for u in adopted.get("unmapped") or [] if u not in unmapped)
     services = token.setdefault("services", [])
     services.extend(s for s in adopted.get("services") or [] if s not in services)
     if services:
@@ -245,35 +344,38 @@ def merge_into(token: dict | None, adopted: dict) -> dict:
     return token
 
 
+def _resume_claimed(claim_path: Path, body: dict) -> None:
+    token = drop_never_stopped(dict(body["token"]))
+    token.update(resume_needed=True, recovery=True)
+    print("→ Restarting gateway(s) paused by an interrupted `hermes update`...", file=sys.stderr)
+    try:
+        from hermes_cli.update_cmd_windows import _resume_windows_gateways_after_update
+        _resume_windows_gateways_after_update(token)
+    except Exception as exc:
+        print(f"  ⚠ Could not restart every paused gateway: {exc}. Run `hermes update` or "
+              "`hermes gateway start`.", file=sys.stderr)
+    finally:
+        with suppress(OSError):
+            if token.get("resume_needed") or token.get("resume_deferred"):
+                # Still owed: hand the claim back unowned so the next launch retries at once — this
+                # launch may live for hours (a chat). A failed rewrite leaves our claimer line, which
+                # turns dead (re-adoptable) when we exit.
+                owed = {key: value for key, value in token.items() if key not in ("recovery", "resume_deferred")}
+                _atomic_write(claim_path, {**body, "token": {**owed, "resume_needed": True}, "claimer": UNOWNED})
+            else:
+                claim_path.unlink()
+
+
 def recover(argv: list[str] | None = None) -> None:
-    """Start-of-run recovery: resume an orphaned pause. Never raises."""
+    """Start-of-run recovery: resume every orphaned pause. Never raises."""
     from hermes_cli._parser import command_argv
     command = command_argv(list(sys.argv[1:] if argv is None else argv))
     if command[:1] == ["update"] or command[:2] == ["gateway", "run"]:
         return  # the update adopts it itself; a booting gateway must not block on its siblings
     try:
-        body = orphaned_record()
-        if body is None:
-            return
-        path = record_path()
-        claim = path.with_name(f"{path.name}.{os.getpid()}.claim")
-        try:
-            os.rename(path, claim)  # exactly one concurrent launch wins the record
-        except OSError:
-            return
-        token = drop_never_stopped(dict(body["token"]))
-        token["resume_needed"] = True
-        print("→ Restarting gateway(s) paused by an interrupted `hermes update`...", file=sys.stderr)
-        try:
-            # Its own finally mirrors the outcome: whatever is still owed is re-recorded
-            # under this process, so the next launch retries once we have exited.
-            from hermes_cli.update_cmd_windows import _resume_windows_gateways_after_update
-            _resume_windows_gateways_after_update(token)
-        except Exception as exc:
-            print(f"  ⚠ Could not restart every paused gateway: {exc}. Run `hermes update` or "
-                  "`hermes gateway start`.", file=sys.stderr)
-        finally:
-            with suppress(OSError):
-                claim.unlink()
+        for src, _body in orphans():
+            won = claim(src)
+            if won is not None:
+                _resume_claimed(*won)
     except Exception as exc:  # never brick a launch on recovery
         print(f"  ⚠ Paused-gateway recovery skipped: {exc}", file=sys.stderr)