fix(compression): prefer psutil.pid_exists for lease liveness probe; add same-pid self-reclaim guard

Hardening on top of the salvaged dead-PID lease reclamation from PR #65775
(@the3asic):

- Probe via psutil.pid_exists (hard dependency; CONTRIBUTING.md critical
  rule #1) with the contributor's os.kill(pid, 0) POSIX probe retained
  only as a scaffold-phase fallback when psutil is missing.
- Same-process holders (pid == os.getpid()) are never probed and never
  self-reclaimed — another thread's live lease is owned by the lease
  refresher/release path.
- Any probe doubt (exceptions, permission errors) conservatively keeps
  the lease until normal TTL expiry; Windows stays TTL-only.
- Tests: psutil-first dead-pid reclaim (probe call pinned), os.kill
  fallback path, probe-doubt keeps lease, same-pid no self-reclaim,
  legacy holder + Windows paths assert NO probe via either API.

hermes_state.py

@@ -31,6 +31,11 @@ from agent.message_sanitization import _sanitize_surrogates
 from hermes_constants import get_hermes_home
 from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar
 
+try:  # Hard dependency, but tolerate scaffold-phase imports before pip install.
+    import psutil
+except ImportError:  # pragma: no cover - stripped/scaffold installs only
+    psutil = None  # type: ignore[assignment]
+
 logger = logging.getLogger(__name__)
 
 _COMPRESSION_LOCK_HOLDER_PID_RE = re.compile(r"(?:^|:)pid=(\d+)(?::|$)")
@@ -44,11 +49,14 @@ def _compression_lock_holder_process_is_dead(holder: str) -> bool:
     process killed during gateway shutdown cannot release its lease, so waiting
     for the full TTL makes every new turn repeatedly attempt compaction. Reclaim
     only when the kernel proves that PID no longer exists; legacy/unstructured
-    holders and permission errors remain protected until normal TTL expiry.
+    holders, same-process holders, permission errors, and any probe doubt
+    remain protected until normal TTL expiry (conservative: PID reuse must
+    never steal a live lease, and a wrongly-kept lease self-heals via TTL).
     """
-    # Python's os.kill(pid, 0) is a non-destructive liveness probe on POSIX.
-    # On Windows, any non-CTRL signal value is implemented with
-    # TerminateProcess, so fall back to TTL-only recovery there.
+    # Windows stays TTL-only: stdlib os.kill(pid, 0) is NOT a no-op probe
+    # there (bpo-14484 — sig=0 maps to CTRL_C_EVENT and can kill the target's
+    # console group), and PID recycling semantics make liveness a weaker
+    # deadness signal. The 300s lease TTL remains the recovery path.
     if os.name == "nt":
         return False
     match = _COMPRESSION_LOCK_HOLDER_PID_RE.search(holder or "")
@@ -60,8 +68,22 @@ def _compression_lock_holder_process_is_dead(holder: str) -> bool:
         return False
     if pid <= 0:
         return False
+    if pid == os.getpid():
+        # Same-process holder (e.g. another thread's live lease): never
+        # self-reclaim — the lease refresher and release path own it.
+        return False
+    if psutil is not None:
+        try:
+            # psutil is the canonical cross-platform liveness answer
+            # (CONTRIBUTING.md "Critical rules" #1). pid_exists() reports
+            # recycled PIDs as alive — conservative, the TTL still applies.
+            return not psutil.pid_exists(pid)
+        except Exception:
+            return False  # any doubt → keep the lease until TTL expiry
+    # Scaffold-phase fallback only (psutil missing). POSIX-only by the
+    # os.name gate above.
     try:
-        os.kill(pid, 0)
+        os.kill(pid, 0)  # windows-footgun: ok — function early-returns on nt above
     except ProcessLookupError:
         return True
     except (PermissionError, OSError, OverflowError):

tests/test_hermes_state_compression_locks.py

@@ -16,6 +16,7 @@ import os
 import threading
 import time
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -109,6 +110,33 @@ def test_non_expired_lock_from_dead_pid_is_reclaimed(
         "sess1", dead_holder, ttl_seconds=300
     ) is True
 
+    probed: list[int] = []
+
+    def process_is_gone(pid: int) -> bool:
+        probed.append(pid)
+        return False
+
+    monkeypatch.setattr(
+        hermes_state, "psutil", SimpleNamespace(pid_exists=process_is_gone)
+    )
+
+    assert db.try_acquire_compression_lock(
+        "sess1", "pid=525252:tid=2:agent=def:nonce=fresh", ttl_seconds=300
+    ) is True
+    assert probed == [424242]
+
+
+def test_dead_pid_reclaim_via_os_kill_fallback_when_psutil_missing(
+    db: SessionDB, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Scaffold-phase installs (no psutil) fall back to os.kill(pid, 0)."""
+    dead_holder = "pid=424242:tid=1:agent=abc:nonce=deadbeef"
+    assert db.try_acquire_compression_lock(
+        "sess1", dead_holder, ttl_seconds=300
+    ) is True
+
+    monkeypatch.setattr(hermes_state, "psutil", None)
+
     def process_is_gone(pid: int, signal: int) -> None:
         assert pid == 424242
         assert signal == 0
@@ -121,6 +149,28 @@ def test_non_expired_lock_from_dead_pid_is_reclaimed(
     ) is True
 
 
+def test_probe_doubt_keeps_lease_until_ttl(
+    db: SessionDB, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A probe that errors out is doubt, not proof of death → TTL protects."""
+    holder = "pid=424242:tid=1:agent=abc:nonce=doubt"
+    assert db.try_acquire_compression_lock(
+        "sess1", holder, ttl_seconds=300
+    ) is True
+
+    def probe_blows_up(pid: int) -> bool:
+        raise RuntimeError("transient probe failure")
+
+    monkeypatch.setattr(
+        hermes_state, "psutil", SimpleNamespace(pid_exists=probe_blows_up)
+    )
+
+    assert db.try_acquire_compression_lock(
+        "sess1", "pid=525252:tid=2:agent=def:nonce=other", ttl_seconds=300
+    ) is False
+    assert db.get_compression_lock_holder("sess1") == holder
+
+
 def test_non_expired_lock_from_live_pid_is_not_reclaimed(db: SessionDB) -> None:
     live_holder = f"pid={os.getpid()}:tid=1:agent=abc:nonce=live"
     assert db.try_acquire_compression_lock(
@@ -131,12 +181,51 @@ def test_non_expired_lock_from_live_pid_is_not_reclaimed(db: SessionDB) -> None:
     ) is False
 
 
+def test_same_process_holder_is_never_self_reclaimed(
+    db: SessionDB, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A holder from THIS pid is never probed — even a lying probe can't steal it."""
+    live_holder = f"pid={os.getpid()}:tid=1:agent=abc:nonce=self"
+    assert db.try_acquire_compression_lock(
+        "sess1", live_holder, ttl_seconds=300
+    ) is True
+    # Even if a (broken) probe were to claim our own PID is dead, the
+    # same-process guard short-circuits before any probe runs.
+    monkeypatch.setattr(
+        hermes_state,
+        "psutil",
+        SimpleNamespace(
+            pid_exists=lambda _pid: pytest.fail(
+                "same-process holder must not be probed"
+            )
+        ),
+    )
+    monkeypatch.setattr(
+        hermes_state.os,
+        "kill",
+        lambda *_args: pytest.fail("same-process holder must not be probed"),
+    )
+    assert db.try_acquire_compression_lock(
+        "sess1", "pid=525252:tid=2:agent=def:nonce=other", ttl_seconds=300
+    ) is False
+    assert db.get_compression_lock_holder("sess1") == live_holder
+
+
 def test_unstructured_holder_waits_for_ttl(
     db: SessionDB, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     assert db.try_acquire_compression_lock(
         "sess1", "legacy_holder", ttl_seconds=300
     ) is True
+    monkeypatch.setattr(
+        hermes_state,
+        "psutil",
+        SimpleNamespace(
+            pid_exists=lambda _pid: pytest.fail(
+                "unstructured holder must not probe a PID"
+            )
+        ),
+    )
     monkeypatch.setattr(
         hermes_state.os,
         "kill",
@@ -147,7 +236,7 @@ def test_unstructured_holder_waits_for_ttl(
     ) is False
 
 
-def test_windows_uses_ttl_only_without_os_kill(
+def test_windows_uses_ttl_only_without_pid_probe(
     db: SessionDB, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     holder = "pid=424242:tid=1:agent=abc:nonce=windows"
@@ -155,6 +244,15 @@ def test_windows_uses_ttl_only_without_os_kill(
         "sess1", holder, ttl_seconds=300
     ) is True
     monkeypatch.setattr(hermes_state.os, "name", "nt")
+    monkeypatch.setattr(
+        hermes_state,
+        "psutil",
+        SimpleNamespace(
+            pid_exists=lambda _pid: pytest.fail(
+                "Windows must stay TTL-only — no PID probe"
+            )
+        ),
+    )
     monkeypatch.setattr(
         hermes_state.os,
         "kill",