fix(update): scope Windows gateway lifecycle to current install

hermes_cli/gateway_windows.py

@@ -992,7 +992,9 @@ def install(
     raise RuntimeError(f"Windows gateway install failed: {detail}")
 
 
-def _live_gateway_pids(all_profiles: bool = False, home: Path | None = None) -> list[int]:
+def _live_gateway_pids(
+    all_profiles: bool = False, home: Path | None = None, pid_filter=None
+) -> list[int]:
     """Live gateway PIDs for the readiness poll. ``home`` scopes the probe to ONE profile's identity
     files (a still-running sibling must not vouch for a per-profile spawn, #110959); otherwise the
     process-table discovery for the active profile or the whole fleet."""
@@ -1000,14 +1002,16 @@ def _live_gateway_pids(all_profiles: bool = False, home: Path | None = None) ->
         from gateway.status import get_running_pid
 
         pid = get_running_pid(home / "gateway.pid", cleanup_stale=False)
-        return [pid] if pid else []
-    from hermes_cli.gateway import find_gateway_pids
-
-    return list(find_gateway_pids(all_profiles=all_profiles))
+        pids = [pid] if pid else []
+    else:
+        from hermes_cli.gateway import find_gateway_pids
+        pids = list(find_gateway_pids(all_profiles=all_profiles))
+    return list(pid_filter(pids)) if pid_filter is not None else pids
 
 
 def _confirm_gateway_stable(
-    initial_pids: list[int], confirm_s: float, interval_s: float, all_profiles: bool = False, home: Path | None = None,
+    initial_pids: list[int], confirm_s: float, interval_s: float, all_profiles: bool = False,
+    home: Path | None = None, pid_filter=None,
 ) -> list[int]:
     """Re-check a freshly detected gateway for ``confirm_s`` seconds: one process-table hit proves
     the child was *created*, not that it survived startup (or a parent Job Object teardown).
@@ -1024,7 +1028,7 @@ def _confirm_gateway_stable(
     confirm_deadline = time.monotonic() + confirm_s
     while time.monotonic() < confirm_deadline:
         time.sleep(interval_s)
-        pids = _live_gateway_pids(all_profiles, home)
+        pids = _live_gateway_pids(all_profiles, home, pid_filter)
         if not pids:
             return []
     return pids
@@ -1032,15 +1036,17 @@ def _confirm_gateway_stable(
 
 def _wait_for_gateway_ready(
     timeout_s: float = 6.0, interval_s: float = 0.4, confirm_s: float = 2.0, all_profiles: bool = False,
-    home: Path | None = None,
+    home: Path | None = None, pid_filter=None,
 ) -> list[int]:
     """Poll for a live gateway for up to ``timeout_s``; a first hit is provisional until the gateway
     stays visible for ``confirm_s`` more seconds (a child that dies right after spawn earns no ✓)."""
     deadline = time.monotonic() + timeout_s
     while time.monotonic() < deadline:
-        pids = _live_gateway_pids(all_profiles, home)
+        pids = _live_gateway_pids(all_profiles, home, pid_filter)
         if pids:
-            confirmed = _confirm_gateway_stable(pids, confirm_s, interval_s, all_profiles=all_profiles, home=home)
+            confirmed = _confirm_gateway_stable(
+                pids, confirm_s, interval_s, all_profiles=all_profiles, home=home, pid_filter=pid_filter
+            )
             if confirmed:
                 return confirmed
             continue  # died during confirmation — keep polling until deadline

hermes_cli/update_cmd_windows.py

@@ -833,6 +833,102 @@ def _pause_windows_gateway_services(service_gateways, token: dict, profiles: dic
         raise RuntimeError(detail) from exc
 
 
+def _gateway_process_install_match(pid: int, project_root: Path, home_root: Path | None) -> bool | None:
+    """True/False for proven current/foreign ownership; None when the process cannot prove either."""
+    psutil = _psutil()
+    if psutil is None:
+        return None
+    try:
+        proc = psutil.Process(int(pid))
+    except Exception:
+        return None
+
+    def _under(raw, root: Path | None) -> bool:
+        if not raw or root is None:
+            return False
+        try:
+            Path(raw).resolve().relative_to(root.resolve())
+            return True
+        except (OSError, RuntimeError, ValueError):
+            return False
+
+    home_match = None
+    try:
+        raw_home = (proc.environ() or {}).get("HERMES_HOME")
+        if raw_home and Path(raw_home).is_absolute():
+            home_match = _under(raw_home, home_root)
+            if home_match:
+                return True
+    except Exception:
+        pass
+
+    try:
+        if _under(proc.exe(), project_root):
+            return True
+    except Exception:
+        pass
+    try:
+        argv = proc.cmdline() or []
+        if argv and _under(argv[0], project_root):
+            return True
+    except Exception:
+        pass
+    return False if home_match is False else None
+
+
+def _classify_current_install_gateway_pids(pids, *, known_owned=()) -> tuple[list[int], list[int]]:
+    """Return (owned, unknown); candidates proven foreign are intentionally omitted."""
+    candidates: list[int] = []
+    for value in pids:
+        try:
+            pid = int(value)
+        except (TypeError, ValueError):
+            continue
+        if pid > 0 and pid not in candidates:
+            candidates.append(pid)
+    if not candidates:
+        return [], []
+
+    candidate_set = set(candidates)
+    owned = {int(pid) for pid in known_owned if int(pid) in candidate_set}
+    from hermes_cli.update_cmd import _m
+    project_root = Path(_m().PROJECT_ROOT)
+
+    try:
+        from hermes_cli.process_identity import ledger_entries
+        for entry in ledger_entries(project_root=project_root, verified_only=True):
+            pid = entry.get("pid")
+            if entry.get("purpose") == "gateway" and isinstance(pid, int) and pid in candidate_set:
+                owned.add(pid)
+    except Exception as exc:
+        logger.debug("Could not read current-install gateway ledger: %s", exc)
+
+    try:
+        from hermes_constants import get_default_hermes_root
+        home_root = Path(get_default_hermes_root())
+    except Exception:
+        home_root = None
+
+    unknown: set[int] = set()
+    for pid in candidates:
+        if pid in owned:
+            continue
+        match = _gateway_process_install_match(pid, project_root, home_root)
+        if match is True:
+            owned.add(pid)
+        elif match is None:
+            unknown.add(pid)
+    return (
+        [pid for pid in candidates if pid in owned],
+        [pid for pid in candidates if pid in unknown],
+    )
+
+
+def _current_install_gateway_pids(pids, *, known_owned=()) -> list[int]:
+    """Owned half of the install-scoped gateway classification, suitable for mutation/readiness."""
+    return _classify_current_install_gateway_pids(pids, known_owned=known_owned)[0]
+
+
 def _discover_windows_gateways():
     """``(profile_processes, service_gateways, service_gateway_pids, running_pids)`` for the pause; any indeterminate probe aborts."""
     from hermes_cli.gateway import find_gateway_pids, find_profile_gateway_processes, find_windows_gateway_services
@@ -843,9 +939,12 @@ def _discover_windows_gateways():
         service_gateways = find_windows_gateway_services(profile_processes=profile_process_list)
     service_gateway_pids = {int(service.gateway_pid) for service in service_gateways}
     with _abort_on_error("Could not discover Windows gateway PIDs before update"):
-        running_pids = list(dict.fromkeys(
+        machine_pids = list(dict.fromkeys(
             [*find_gateway_pids(all_profiles=True), *sorted(profile_processes), *sorted(service_gateway_pids)]
         ))
+        running_pids = _current_install_gateway_pids(
+            machine_pids, known_owned=set(profile_processes) | service_gateway_pids
+        )
     return profile_processes, service_gateways, service_gateway_pids, running_pids
 
 
@@ -1037,7 +1136,16 @@ def _cold_start_windows_gateway_after_update(token: dict | None = None) -> bool:
         from hermes_cli import gateway_windows
         from hermes_cli.gateway import find_gateway_pids
     with _abort_on_error("Could not re-check gateway liveness before cold-start"):
-        if list(find_gateway_pids(all_profiles=True)):
+        owned_pids, unknown_pids = _classify_current_install_gateway_pids(
+            find_gateway_pids(all_profiles=True)
+        )
+        if owned_pids:
+            return True
+        if unknown_pids:
+            logger.debug(
+                "Skipping Windows gateway cold-start: ownership is indeterminate for PID(s) %s",
+                ", ".join(map(str, unknown_pids)),
+            )
             return True
     token = token or {}
     generation = token.get("attested_generation")
@@ -1252,7 +1360,9 @@ def _verify_relaunched_gateways_alive(token: dict, profiles: dict, unmapped: lis
         from gateway.status import _pid_exists
         from hermes_cli import gateway_windows
     timeout_s = _relaunch_verify_timeout_s(profiles, unmapped, _pid_exists)
-    ready_pids = gateway_windows._wait_for_gateway_ready(timeout_s=timeout_s, all_profiles=True)
+    ready_pids = gateway_windows._wait_for_gateway_ready(
+        timeout_s=timeout_s, all_profiles=True, pid_filter=_current_install_gateway_pids
+    )
     if not ready_pids:
         token["profiles"] = dict(profiles)
         token["unmapped"] = list(unmapped)

tests/hermes_cli/test_update_windows_cross_install_gateway.py

@@ -0,0 +1,176 @@
+"""Windows update gateway ownership is install-scoped, not machine-scoped (#124659)."""
+
+from types import SimpleNamespace
+
+from hermes_cli import gateway as gateway_mod
+from hermes_cli import gateway_windows
+from hermes_cli import main as cli_main
+import hermes_cli.main_install_repair as main_install_repair
+from hermes_cli import process_identity
+from hermes_cli import update_cmd
+from hermes_cli import update_cmd_windows
+
+
+def test_current_install_filter_keeps_ledger_owned_unmapped_gateway(monkeypatch, tmp_path):
+    root = tmp_path / "checkout"
+    home = tmp_path / "home"
+    root.mkdir()
+    home.mkdir()
+    monkeypatch.setattr(cli_main, "PROJECT_ROOT", root)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(
+        process_identity, "ledger_entries", lambda **_kwargs: [{"pid": 101, "purpose": "gateway"}]
+    )
+    monkeypatch.setattr(update_cmd_windows, "_psutil", lambda: None)
+
+    assert update_cmd_windows._current_install_gateway_pids([101, 202]) == [101]
+
+
+def test_legacy_home_proves_owned_unmapped_gateway(monkeypatch, tmp_path):
+    root = tmp_path / "checkout"
+    home = tmp_path / "hermes"
+    foreign = tmp_path / "foreign"
+    for path in (root, home, foreign):
+        path.mkdir()
+    profile_home = home / "profiles" / "work"
+    profile_home.mkdir(parents=True)
+    monkeypatch.setattr(cli_main, "PROJECT_ROOT", root)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(process_identity, "ledger_entries", lambda **_kwargs: [])
+
+    class Proc:
+        def __init__(self, pid):
+            self.pid = pid
+
+        def environ(self):
+            return {"HERMES_HOME": str(profile_home if self.pid == 101 else foreign)}
+
+        def exe(self):
+            return str(foreign / "python.exe")
+
+        def cmdline(self):
+            return [str(foreign / "python.exe")]
+
+    monkeypatch.setattr(update_cmd_windows, "_psutil", lambda: SimpleNamespace(Process=Proc))
+
+    owned, unknown = update_cmd_windows._classify_current_install_gateway_pids([101, 202])
+    assert owned == [101]
+    assert unknown == []
+
+
+def test_cwd_alone_is_not_destructive_ownership(monkeypatch, tmp_path):
+    root = tmp_path / "checkout"
+    home = tmp_path / "hermes"
+    foreign = tmp_path / "foreign"
+    for path in (root, home, foreign):
+        path.mkdir()
+    monkeypatch.setattr(cli_main, "PROJECT_ROOT", root)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(process_identity, "ledger_entries", lambda **_kwargs: [])
+
+    proc = SimpleNamespace(
+        environ=lambda: {},
+        exe=lambda: str(foreign / "python.exe"),
+        cmdline=lambda: [str(foreign / "python.exe")],
+        cwd=lambda: str(root),
+    )
+    monkeypatch.setattr(
+        update_cmd_windows, "_psutil", lambda: SimpleNamespace(Process=lambda _pid: proc)
+    )
+
+    owned, unknown = update_cmd_windows._classify_current_install_gateway_pids([202])
+    assert owned == []
+    assert unknown == [202]
+
+
+def test_discovery_keeps_owned_unmapped_and_excludes_foreign(monkeypatch):
+    monkeypatch.setattr(gateway_mod, "find_profile_gateway_processes", lambda **_kwargs: [])
+    monkeypatch.setattr(gateway_mod, "find_windows_gateway_services", lambda **_kwargs: [])
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_kwargs: [101, 202])
+    monkeypatch.setattr(
+        update_cmd_windows,
+        "_current_install_gateway_pids",
+        lambda pids, **_kwargs: [pid for pid in pids if pid == 101],
+    )
+
+    profiles, services, service_pids, running = update_cmd_windows._discover_windows_gateways()
+
+    assert profiles == {}
+    assert services == []
+    assert service_pids == set()
+    assert running == [101]
+
+
+def test_foreign_gateway_does_not_suppress_current_install_cold_start(monkeypatch):
+    monkeypatch.setattr(cli_main, "_is_windows", lambda: True)
+    monkeypatch.setattr(main_install_repair, "_is_windows", lambda: True)
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_kwargs: [202])
+    monkeypatch.setattr(
+        update_cmd_windows,
+        "_classify_current_install_gateway_pids",
+        lambda _pids, **_kwargs: ([], []),
+    )
+    monkeypatch.setattr(update_cmd, "_desktop_owns_gateway_lifecycle", lambda: False)
+
+    spawned = []
+    monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda: spawned.append(True) or 4242)
+    monkeypatch.setattr(gateway_windows, "_wait_for_gateway_ready", lambda *a, **k: [4242])
+    monkeypatch.setattr(gateway_windows, "_write_start_attestation", lambda *a, **k: None)
+
+    assert update_cmd_windows._cold_start_windows_gateway_after_update({"attested_generation": None})
+    assert spawned == [True]
+
+
+def test_unknown_gateway_conservatively_suppresses_cold_start(monkeypatch):
+    monkeypatch.setattr(cli_main, "_is_windows", lambda: True)
+    monkeypatch.setattr(main_install_repair, "_is_windows", lambda: True)
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_kwargs: [303])
+    monkeypatch.setattr(
+        update_cmd_windows,
+        "_classify_current_install_gateway_pids",
+        lambda _pids, **_kwargs: ([], [303]),
+    )
+    monkeypatch.setattr(update_cmd, "_desktop_owns_gateway_lifecycle", lambda: False)
+    monkeypatch.setattr(
+        gateway_windows,
+        "_spawn_detached",
+        lambda: (_ for _ in ()).throw(AssertionError("must not spawn with unknown ownership")),
+    )
+
+    assert update_cmd_windows._cold_start_windows_gateway_after_update(
+        {"attested_generation": None}
+    )
+
+
+def test_readiness_filter_waits_past_foreign_gateway(monkeypatch):
+    snapshots = iter(([202], [202, 4242]))
+    monkeypatch.setattr(
+        gateway_mod, "find_gateway_pids", lambda **_kwargs: next(snapshots, [202, 4242])
+    )
+    monkeypatch.setattr(gateway_windows.time, "sleep", lambda _seconds: None)
+
+    ready = gateway_windows._wait_for_gateway_ready(
+        timeout_s=1.0,
+        confirm_s=0,
+        all_profiles=True,
+        pid_filter=lambda pids: [pid for pid in pids if pid == 4242],
+    )
+
+    assert ready == [4242]
+
+
+def test_post_relaunch_verification_uses_install_filter(monkeypatch):
+    captured = {}
+
+    def wait_for_ready(**kwargs):
+        captured.update(kwargs)
+        return [4242]
+
+    monkeypatch.setattr(gateway_windows, "_wait_for_gateway_ready", wait_for_ready)
+    monkeypatch.setattr(gateway_windows, "_write_start_attestation", lambda *a, **k: None)
+    monkeypatch.setattr(update_cmd_windows, "_relaunch_verify_timeout_s", lambda *_args: 1.0)
+
+    update_cmd_windows._verify_relaunched_gateways_alive({}, {}, [])
+
+    assert captured["all_profiles"] is True
+    assert captured["pid_filter"] is update_cmd_windows._current_install_gateway_pids