fix(sessions): skip POSIX zombie probe on Windows; prune registry snapshot after lock release

_pid_exists() ran psutil.Process(pid).status() (~7 ms) unconditionally before the
Windows branch — once per registry entry inside the unfair exclusive session file
lock, starving every 2 Hz poller at >=7 leases (#115578). Windows has no zombie
state, so the probe is pure cost there; pid_exists()/ctypes decide instead.

active_session_registry_snapshot() read AND pruned under _FileLock. It now snapshots
raw entries under the lock, probes liveness after release, and re-locks only to drop
the lease ids proven dead, so a lease acquired in between survives.

Tests trimmed from the contributor branch to two invariants + a POSIX control; the
Windows half is proven on the real runner via the wine2e red/green receipt.

Salvages #115591

gateway/status.py

@@ -773,6 +773,10 @@ def _pid_exists(pid: int) -> bool:
     try:
         import psutil  # type: ignore
         # Best-effort zombie check: status-read failures fall through to pid_exists().
+        # Windows has no POSIX zombies, and this probe costs ~7 ms per call — once per
+        # registry entry inside the session file lock (#115578). Skip it on Windows and
+        # let pid_exists() below (or the ctypes fallback) decide.
+        probe_zombie = os.name != "nt"
         try:
             # A zombie (defunct) process is still in the process table, so ``psutil.pid_exists()`` returns
             # True for it — but it is already dead: SIGKILL has no effect and it cannot be a running
@@ -782,7 +786,7 @@ def _pid_exists(pid: int) -> bool:
             # #42126). Report zombies as dead so the takeover proceeds. Best-effort: any failure to read
             # status (partial/stub psutil, access denied, transient race) falls through to the authoritative
             # ``pid_exists()`` below rather than raising.
-            if psutil.Process(pid).status() == psutil.STATUS_ZOMBIE:
+            if probe_zombie and psutil.Process(pid).status() == psutil.STATUS_ZOMBIE:
                 return False
         except getattr(psutil, "NoSuchProcess", ()):
             return False

hermes_cli/active_sessions.py

@@ -711,14 +711,35 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int:
 def active_session_registry_snapshot(
     registry_home: str | Path | None = None, *, strict: bool = False,
 ) -> list[dict[str, Any]]:
-    """Return live leases; attachment callers require provable liveness."""
+    """Return live leases; attachment callers require provable liveness.
+
+    The per-entry liveness probes in ``_prune_dead`` run AFTER the file lock
+    is released: holding an exclusive, unfair lock across process-introspection
+    syscalls starves concurrent pollers once a handful of leases exist
+    (#115578). The prune write-back re-locks and drops only the lease ids
+    already proven dead, so a lease created between the snapshot and the
+    write-back is never lost.
+    """
     state_path, lock_path = _lease_paths(registry_home=registry_home)
     with _FileLock(lock_path):
         raw_entries = _read_entries(state_path, strict=True)
-        entries = _prune_dead(raw_entries, strict=strict)
-        if entries != raw_entries:
-            _write_entries(state_path, entries)
-        return entries
+    entries = _prune_dead(raw_entries, strict=strict)
+    if entries != raw_entries:
+        live_lease_ids = {str(entry.get("lease_id") or "") for entry in entries}
+        dead_lease_ids = {
+            str(entry.get("lease_id") or "")
+            for entry in raw_entries
+            if str(entry.get("lease_id") or "") not in live_lease_ids
+        }
+        with _FileLock(lock_path):
+            current_entries = _read_entries(state_path, strict=True)
+            kept_entries = [
+                entry for entry in current_entries
+                if str(entry.get("lease_id") or "") not in dead_lease_ids
+            ]
+            if len(kept_entries) != len(current_entries):
+                _write_entries(state_path, kept_entries)
+    return entries
 
 
 @contextmanager

tests/gateway/test_status.py

@@ -633,6 +633,40 @@ class TestTerminatePid:
         assert calls == []
 
 
+class TestPidExistsZombieProbe:
+    """#115578: the psutil ``status()`` zombie probe is POSIX-only. On Windows it costs ~7 ms per
+    pid, runs once per registry entry inside the session-registry file lock, and can never
+    report a zombie (Windows has no such state), so pollers starve at a handful of leases."""
+
+    @staticmethod
+    def _spy_status(monkeypatch):
+        psutil = pytest.importorskip("psutil")
+        calls = []
+        real_status = psutil.Process.status
+
+        def spy(self):
+            calls.append(self.pid)
+            return real_status(self)
+
+        monkeypatch.setattr(psutil.Process, "status", spy)
+        return calls
+
+    @pytest.mark.windows_only
+    def test_windows_skips_zombie_status_probe(self, monkeypatch):
+        # Faking os.name on POSIX proves nothing about the cost on the real host; the wine2e
+        # runner receipt (red on main, green on the fix) is the live repro for this test.
+        calls = self._spy_status(monkeypatch)
+        assert status._pid_exists(os.getpid()) is True
+        assert calls == []
+
+    @pytest.mark.linux_only
+    def test_posix_still_probes_zombie_status(self, monkeypatch):
+        # Control: on POSIX a zombie still answers pid_exists(), so the probe must survive.
+        calls = self._spy_status(monkeypatch)
+        assert status._pid_exists(os.getpid()) is True
+        assert calls == [os.getpid()]
+
+
 class TestScopedLocks:
     @pytest.mark.windows_only
     def test_windows_file_lock_uses_high_offset(self, tmp_path, monkeypatch):

tests/hermes_cli/test_active_sessions.py

@@ -741,3 +741,44 @@ def test_pid_liveness_self_pid_skips_exists_probe(monkeypatch):
     # Identity is still (pid, start time): our pid with a start we never had is a recycled pid.
     assert active_sessions._pid_liveness(os.getpid(), 1.0) is False
     assert exists_calls == []
+
+
+def test_snapshot_prunes_after_lock_release_and_keeps_concurrent_lease(tmp_path, monkeypatch):
+    """#115578: ``active_session_registry_snapshot`` must not run the per-entry liveness
+    probes while holding the exclusive registry lock (one psutil round-trip per lease held
+    under an unfair ``LK_LOCK`` starves every 2 Hz poller past ~7 leases), and the pruned
+    write-back must not drop a lease acquired between the snapshot and the write."""
+    home = tmp_path / ".hermes"
+    state_path = home / "runtime" / "active_sessions.json"
+    lock_path = home / "runtime" / "active_sessions.lock"
+    state_path.parent.mkdir(parents=True)
+    dead_pid = 2**30  # never a live pid: _pid_exists() is False, nothing is signalled
+    active_sessions._write_entries(state_path, [
+        {"lease_id": "live-1", "session_id": "s-live", "pid": os.getpid()},
+        {"lease_id": "dead-1", "session_id": "s-dead", "pid": dead_pid},
+    ])
+    newcomer = {"lease_id": "new-1", "session_id": "s-new", "pid": os.getpid()}
+    real_prune = active_sessions._prune_dead
+    acquired_during_prune = []
+
+    def concurrent_acquire():
+        with active_sessions._FileLock(lock_path):
+            current = active_sessions._read_entries(state_path, strict=True)
+            active_sessions._write_entries(state_path, current + [dict(newcomer)])
+
+    def spy_prune(entries, **kwargs):
+        # A concurrent acquirer must be able to take the lock while the probes run.
+        acquirer = threading.Thread(target=concurrent_acquire, daemon=True)
+        acquirer.start()
+        acquirer.join(timeout=5)
+        acquired_during_prune.append(not acquirer.is_alive())
+        return real_prune(entries, **kwargs)
+
+    monkeypatch.setattr(active_sessions, "_prune_dead", spy_prune)
+
+    live = active_sessions.active_session_registry_snapshot(registry_home=home)
+
+    assert acquired_during_prune == [True]
+    assert [e["lease_id"] for e in live] == ["live-1"]
+    persisted = active_sessions._read_entries(state_path, strict=True)
+    assert sorted(e["lease_id"] for e in persisted) == ["live-1", "new-1"]