fix: re-attach a live recovered PID instead of inventing an exit

Checkpoint refresh treated a failed start-time check as a collected exit
and queued a completion. A live PID whose start time matches, or whose
start time cannot be read, stays running. A reused PID is closed without
being signalled. A gone PID is pruned. A completion is emitted only when
an exit status was collected.

tests/tools/test_process_checkpoint_readopt.py

@@ -0,0 +1,196 @@
+"""Checkpoint recovery must not invent an exit for a live child.
+
+A still-live host PID is re-attached and stays running. A confirmed start-time
+mismatch means the PID was reused: the old entry is closed and that PID is not
+signalled. A gone PID is pruned. A completion is queued only when a real exit
+status was collected.
+"""
+
+import json
+import subprocess
+import sys
+import time
+
+import pytest
+
+from tools.process_registry import ProcessRegistry, ProcessSession
+
+
+@pytest.fixture()
+def registry():
+    return ProcessRegistry()
+
+
+def _sleep_proc() -> subprocess.Popen:
+    return subprocess.Popen(
+        [sys.executable, "-c", "import time; time.sleep(30)"],
+        stdin=subprocess.DEVNULL,
+    )
+
+
+def _detach(registry, proc, *, sid, start, notify=True) -> ProcessSession:
+    session = ProcessSession(
+        id=sid,
+        command="sleep-for-readopt",
+        task_id="t1",
+        started_at=time.time(),
+        pid=proc.pid,
+        pid_scope="host",
+        detached=True,
+        host_start_time=start,
+        notify_on_complete=notify,
+    )
+    registry._running[session.id] = session
+    return session
+
+
+def _completions(registry) -> list:
+    found = []
+    while not registry.completion_queue.empty():
+        found.append(registry.completion_queue.get_nowait())
+    return found
+
+
+class TestCheckpointReadopt:
+    def test_live_matching_pid_stays_running_when_later_start_probe_fails(
+        self, registry, monkeypatch
+    ):
+        """A recovered child that is still alive must not be reported exited
+        just because the start-time probe cannot be read."""
+        proc = _sleep_proc()
+        try:
+            start = ProcessRegistry._safe_host_start_time(proc.pid)
+            assert start is not None
+            session = _detach(registry, proc, sid="proc_live_probe", start=start)
+            monkeypatch.setattr(ProcessRegistry, "_safe_host_start_time", staticmethod(lambda _pid: None))
+
+            polled = registry.poll(session.id)
+
+            assert proc.poll() is None
+            assert polled["status"] == "running"
+            assert "exit_code" not in polled
+            assert session.exited is False
+            assert session.id in registry._running
+            assert _completions(registry) == []
+        finally:
+            proc.kill()
+            proc.wait()
+
+    def test_reused_pid_is_closed_without_kill_or_completion(self, registry):
+        """A live PID whose start time does not match was reused. Close the
+        old entry, do not signal that PID, and do not emit a completion."""
+        proc = _sleep_proc()
+        try:
+            real = ProcessRegistry._safe_host_start_time(proc.pid)
+            assert real is not None
+            session = _detach(registry, proc, sid="proc_reused", start=real + 1)
+
+            polled = registry.poll(session.id)
+
+            assert proc.poll() is None, "reused PID must not be killed"
+            assert session.exited is True
+            assert session.exit_code is None
+            assert session.id in registry._finished
+            assert session.id not in registry._running
+            assert _completions(registry) == []
+            assert polled["status"] != "running"
+            assert polled.get("exit_code") is None
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait()
+
+    def test_gone_pid_is_pruned_without_a_false_completion(self, registry):
+        session = ProcessSession(
+            id="proc_gone",
+            command="already-dead",
+            task_id="t1",
+            started_at=time.time(),
+            pid=2**31 - 1,
+            pid_scope="host",
+            detached=True,
+            host_start_time=1,
+            notify_on_complete=True,
+        )
+        registry._running[session.id] = session
+
+        listed = registry.list_sessions()
+        polled = registry.poll(session.id)
+
+        assert session.id not in registry._running
+        assert all(row["session_id"] != session.id for row in listed)
+        assert polled["status"] == "not_found"
+        assert _completions(registry) == []
+
+    def test_recover_reattaches_live_pid_when_start_probe_fails(
+        self, registry, tmp_path, monkeypatch
+    ):
+        proc = _sleep_proc()
+        try:
+            start = ProcessRegistry._safe_host_start_time(proc.pid)
+            assert start is not None
+            checkpoint = tmp_path / "procs.json"
+            checkpoint.write_text(json.dumps([{
+                "session_id": "proc_recover_live",
+                "command": "sleep-for-readopt",
+                "pid": proc.pid,
+                "pid_scope": "host",
+                "host_start_time": start,
+                "task_id": "t1",
+                "notify_on_complete": True,
+            }]))
+            monkeypatch.setattr(
+                "tools.process_registry.CHECKPOINT_PATH", checkpoint
+            )
+            monkeypatch.setattr(
+                ProcessRegistry, "_safe_host_start_time", staticmethod(lambda _pid: None)
+            )
+
+            recovered = registry.recover_from_checkpoint()
+            polled = registry.poll("proc_recover_live")
+
+            assert recovered == 1
+            assert proc.poll() is None
+            assert polled["status"] == "running"
+            assert "exit_code" not in polled
+            assert _completions(registry) == []
+        finally:
+            proc.kill()
+            proc.wait()
+
+    def test_recover_does_not_kill_a_reused_pid(self, registry, tmp_path, monkeypatch):
+        proc = _sleep_proc()
+        try:
+            real = ProcessRegistry._safe_host_start_time(proc.pid)
+            assert real is not None
+            checkpoint = tmp_path / "procs.json"
+            checkpoint.write_text(json.dumps([{
+                "session_id": "proc_recover_reused",
+                "command": "sleep-for-readopt",
+                "pid": proc.pid,
+                "pid_scope": "host",
+                "host_start_time": real + 1,
+                "task_id": "t1",
+                "notify_on_complete": True,
+                "systemd_unit": "hermes-worker-proc_recover_reused.scope",
+            }]))
+            stopped = []
+            monkeypatch.setattr(
+                "tools.process_registry.CHECKPOINT_PATH", checkpoint
+            )
+            monkeypatch.setattr(
+                "tools.process_registry._stop_systemd_unit",
+                lambda unit: stopped.append(unit) or True,
+            )
+
+            recovered = registry.recover_from_checkpoint()
+
+            assert recovered == 0
+            assert proc.poll() is None
+            assert registry.get("proc_recover_reused") is None
+            assert stopped == []
+            assert _completions(registry) == []
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait()

tools/process_registry.py

@@ -915,21 +915,85 @@ class ProcessRegistry(ProcessCheckpointMixin):
         return cls._is_host_pid_alive(pid) and (
             expected_start is None or cls._safe_host_start_time(pid) == expected_start)
 
+    def _detached_host_fate(self, pid: Optional[int], expected_start: Optional[int]) -> str:
+        """How a recovered host PID should be supervised.
+
+        ``running`` — alive and still ours (start time matches, no baseline, or
+        the start-time probe could not be read). Re-attach; do not invent an exit.
+        ``reused`` — alive, and the start time positively differs. The number
+        belongs to someone else; close our entry and never signal that PID.
+        ``gone`` — the PID is not alive. Prune it; no exit status was collected.
+        """
+        if self._host_pid_is_ours(pid, expected_start):
+            return "running"
+        if not pid or not self._is_host_pid_alive(pid):
+            return "gone"
+        if expected_start is None:
+            return "running"
+        current = self._safe_host_start_time(pid)
+        if current is None or current == expected_start:
+            return "running"
+        return "reused"
+
     def _refresh_detached_session(self, session: Optional[ProcessSession]) -> Optional[ProcessSession]:
-        """Update recovered host-PID sessions when the underlying process has exited."""
+        """Re-attach, close, or prune a recovered host-PID session.
+
+        A completion is not queued here: recovery has no waitable handle, so it
+        never collected an exit status.
+        """
         if session is None or session.exited or not session.detached or session.pid_scope != "host":
             return session
-        # A recycled PID (alive but not ours) counts as "our process exited" so a
-        # later kill() can never tree-kill the stranger.
-        if self._host_pid_is_ours(session.pid, session.host_start_time):
+        fate = self._detached_host_fate(session.pid, session.host_start_time)
+        if fate == "running":
             return session
+        if fate == "gone":
+            return self._prune_uncollected_detached(session)
+        self._close_reused_detached(session)
+        return session
+
+    def _close_reused_detached(self, session: ProcessSession) -> None:
+        """Close an entry whose PID was recycled onto another process.
+
+        The stranger is not signalled, and no completion is queued: we never
+        collected an exit status for the process we spawned.
+        """
         with session._lock:
             if session.exited:
-                return session
-            # No waitable handle survives recovery, so the real exit code is unknown.
-            session.exited, session.exit_code = True, None
-        self._move_to_finished(session)
-        return session
+                return
+            session.exited = True
+        with self._lock:
+            if session.id in self._running:
+                session.exited_at = time.time()
+                self._running.pop(session.id, None)
+            self._finished[session.id] = session
+        self._write_checkpoint()
+        session._completion_event.set()
+
+    def _prune_uncollected_detached(self, session: ProcessSession) -> Optional[ProcessSession]:
+        """Drop a recovered entry whose PID is gone, without inventing an exit.
+
+        An owned systemd scope stays reachable so kill can still reap it. A
+        scope-less entry is removed: poll/list must not report a collected exit.
+        """
+        with session._lock:
+            session.exited = True
+        with self._lock:
+            self._running.pop(session.id, None)
+            if session.systemd_unit:
+                self._finished[session.id] = session
+            else:
+                self._finished.pop(session.id, None)
+        self._write_checkpoint()
+        session._completion_event.set()
+        return session if session.systemd_unit else None
+
+    def _uncollected_gone(self, session: Optional[ProcessSession]) -> bool:
+        """True when a detached entry was closed without a collected exit status
+        and the PID is no longer alive. List/poll must not report that as exited."""
+        return bool(
+            session is not None and session.exited and session.exit_code is None
+            and session.detached and session.pid_scope == "host"
+            and not self._is_host_pid_alive(session.pid))
 
     @staticmethod
     def _proc_alive(proc) -> bool:
@@ -1705,7 +1769,10 @@ class ProcessRegistry(ProcessCheckpointMixin):
         if session is None:
             return False
         with suppress(Exception):
-            self._refresh_detached_session(session)
+            refreshed = self._refresh_detached_session(session)
+            if refreshed is None:
+                return False
+            session = refreshed
         return not session.exited and not (
             session.watch_patterns and not session._watch_disabled and session._watch_hits > 0)
 
@@ -1774,7 +1841,8 @@ class ProcessRegistry(ProcessCheckpointMixin):
                         # where the reader is blocked but the direct child has already exited (issue
                         # #17327).
                         self._reconcile_local_exit(session)
-                        self._refresh_detached_session(session)
+                        if self._refresh_detached_session(session) is None:
+                            break
                     if session._completion_event.is_set():
                         break
                     session._completion_event.wait(min(remaining, interval))
@@ -1996,7 +2064,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
     def poll(self, session_id: str) -> dict:
         """Check status and get new output for a background process."""
         session = self.get(session_id)
-        if session is None:
+        if session is None or self._uncollected_gone(session):
             return _not_found(session_id)
         self._reconcile_local_exit(session)  # orphaned-pipe reader guard
         with session._lock:
@@ -2232,12 +2300,13 @@ class ProcessRegistry(ProcessCheckpointMixin):
                 if session.systemd_unit:
                     _stop_systemd_unit(session.systemd_unit)
                 with session._lock:
-                    session.exited = True
-                    session.exit_code = None
                     output = _completion_output(session)
                 if consume_output:
                     self._completion_consumed.add(session_id)
-                self._move_to_finished(session)
+                # No waitable handle, so this is not a collected exit. Close the
+                # entry without queueing a completion, and do not signal a PID
+                # whose start time does not match.
+                self._close_reused_detached(session)
                 return {"status": "already_exited", "exit_code": session.exit_code, **output}
             self._terminate_host_pid(session.pid, session.host_start_time)
         else:
@@ -2342,7 +2411,11 @@ class ProcessRegistry(ProcessCheckpointMixin):
         with self._lock:
             sessions.update(self._finished)
             sessions.update(self._running)
-        all_sessions = [self._refresh_detached_session(s) for s in sessions.values()]
+        all_sessions = [
+            refreshed for refreshed in (
+                self._refresh_detached_session(s) for s in sessions.values()
+            ) if refreshed is not None and not self._uncollected_gone(refreshed)
+        ]
         if task_id or session_key:
             all_sessions = [
                 s for s in all_sessions
@@ -2412,6 +2485,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
         with self._lock:
             return [s for s in self._finished.values()
                     if s.owner_task_id == owner_task_id and s.notify_on_complete
+                    and s.exit_code is not None
                     and s.id not in self._completion_consumed and s.id not in self._poll_observed]
 
     def transfer_ownership(self, session_id: str, *, from_owner: str, to_owner: str, to_task_id: str,

tools/process_registry_checkpoint.py

@@ -11,6 +11,10 @@ logger = logging.getLogger("tools.process_registry")
 
 
 class ProcessCheckpointMixin:
+    def _detached_host_fate(self, pid: Optional[int], expected_start: Optional[int]) -> str:
+        """Subclass supplies the live PID decision. See ProcessRegistry."""
+        raise NotImplementedError
+
     # ----- Checkpoint (crash recovery) -----
 
     def _write_checkpoint(self, extra_entries: Optional[List[Dict[str, Any]]] = None):
@@ -75,16 +79,20 @@ class ProcessCheckpointMixin:
                     "Skipping recovery for non-host process: %s (pid=%s, scope=%s)",
                     entry.get("command", "unknown")[:60], pid, pid_scope)
                 continue
-            # Alive AND the same process: across a restart the kernel may have
-            # recycled the PID onto a stranger, and adopting it would let a later
-            # kill tree-kill e.g. a browser.
-            if not self._host_pid_is_ours(pid, entry.get("host_start_time")):
-                if self._is_host_pid_alive(pid):
-                    logger.info(
-                        "Not recovering session %s: pid %d is alive but its "
-                        "start time no longer matches — PID was recycled onto "
-                        "an unrelated process; refusing to adopt it.",
-                        entry.get("session_id", "?"), pid)
+            # Alive and still ours: re-attach. A start-time probe that cannot
+            # be read is not proof the PID was reused — dropping it would leave
+            # a live child unsupervised, and marking it exited would invent a
+            # completion. A positive mismatch means the number was recycled:
+            # do not adopt it and do not signal it.
+            fate = self._detached_host_fate(pid, entry.get("host_start_time"))
+            if fate == "reused":
+                logger.info(
+                    "Not recovering session %s: pid %d is alive but its "
+                    "start time no longer matches — PID was recycled onto "
+                    "an unrelated process; refusing to adopt or signal it.",
+                    entry.get("session_id", "?"), pid)
+                continue
+            if fate != "running":
                 systemd_unit = entry.get("systemd_unit", "")
                 if systemd_unit and not _stop_systemd_unit(systemd_unit):
                     logger.warning(