diff --git a/hermes_cli/update_cmd_windows.py b/hermes_cli/update_cmd_windows.py index bcef081c67..2c36161482 100644 --- a/hermes_cli/update_cmd_windows.py +++ b/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: diff --git a/hermes_cli/update_lock.py b/hermes_cli/update_lock.py index dd59b6eb92..725e2ab13a 100644 --- a/hermes_cli/update_lock.py +++ b/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: diff --git a/hermes_cli/update_pause_record.py b/hermes_cli/update_pause_record.py index 8fd60dfeb6..3f0e348034 100644 --- a/hermes_cli/update_pause_record.py +++ b/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 ``...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)