diff --git a/tests/tools/test_process_checkpoint_readopt.py b/tests/tools/test_process_checkpoint_readopt.py new file mode 100644 index 0000000000..3309ea3c90 --- /dev/null +++ b/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() diff --git a/tools/process_registry.py b/tools/process_registry.py index 89a2f360ef..88316d1459 100644 --- a/tools/process_registry.py +++ b/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, diff --git a/tools/process_registry_checkpoint.py b/tools/process_registry_checkpoint.py index 04a140b1e2..7a45491342 100644 --- a/tools/process_registry_checkpoint.py +++ b/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(