fix(process): a completion is published once, by its owner, with its final output

Child exit, result finalization, notice publication and output consumption are
separate steps, but several callers treated an earlier one as the last:

- A duplicate _move_to_finished set _completion_event while the owning call was
  still before its notice, so a one-shot parent could exit without the follow-up.
  Only the call that moved the session out of _running releases waiters now, and
  it always does, even when its publish raises.
- _finish_exited ran mark_exited without the session lock, so a kill committed
  between its `killed` check and its writes was overwritten with exited/0.
- After a reconcile flips `exited` ahead of a reader still on its final drain,
  wait/read_log/kill no longer consume the partial output as the completion,
  and poll no longer marks it observed. poll/read_log/kill decide that under
  the same lock hold as their snapshot, so a reader finishing in between cannot
  make an earlier partial snapshot count as the whole output.
- kill_process ignored a failed systemd scope stop. It now returns an error
  with scope_stop_failed whenever the stop fails, including when the root
  exited meanwhile (that proves nothing about detached descendants); the
  session keeps its unit so a later kill retries the stop. An already-exited
  session reports scope_stop_failed too.

Process-tree termination and post-kill survivor checks move into
tools/process_registry_termination.py (ProcessTerminationMixin) to keep
process_registry.py under its size cap. The only change in the move:
_live_descendants logs the traceback of the failure it swallows.

Refs #132358. The races were found by @andrexibiza and @jonpol01 in that thread.

tests/tools/test_process_registry_completion_races.py

@@ -0,0 +1,242 @@
+"""A process's completion is published once, by its owner, and only with its final output.
+
+Child exit, result finalization, notice publication and output consumption are separate steps.
+Each test forces one interleaving where another caller treated an earlier step as the last one.
+"""
+
+import shlex
+import sys
+import threading
+import time
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+
+import tools.process_registry as module
+from tools.process_registry import ProcessRegistry, ProcessSession
+
+
+def _session(sid: str, **fields) -> ProcessSession:
+    return ProcessSession(id=sid, command="cmd", task_id="task", started_at=time.time(), **fields)
+
+
+def test_duplicate_finisher_leaves_completion_to_the_owner(monkeypatch):
+    """A second finisher must not release waiters while the first is still publishing the notice."""
+    registry = ProcessRegistry()
+    session = _session("proc_dup_finish", notify_on_complete=True)
+    registry._running[session.id] = session
+    entered, release = threading.Event(), threading.Event()
+    calls = []
+
+    def checkpoint(*_args, **_kwargs):
+        calls.append(None)
+        if len(calls) == 1:  # the owner, between the registry move and the notice
+            entered.set()
+            release.wait(10)
+
+    monkeypatch.setattr(registry, "_write_checkpoint", checkpoint)
+    owner = threading.Thread(target=registry._move_to_finished, args=(session,), daemon=True)
+    owner.start()
+    try:
+        assert entered.wait(5)
+        assert registry._move_to_finished(session) is False
+        assert not session._completion_event.is_set(), "duplicate released waiters before the notice"
+        assert registry.completion_queue.empty()
+    finally:
+        release.set()
+        owner.join(5)
+    assert session._completion_event.is_set()
+    assert registry.completion_queue.qsize() == 1
+
+
+def test_owner_that_fails_mid_publish_still_releases_waiters(monkeypatch):
+    registry = ProcessRegistry()
+    session = _session("proc_owner_fails", notify_on_complete=True)
+    registry._running[session.id] = session
+
+    def checkpoint(*_args, **_kwargs):
+        raise RuntimeError("checkpoint write failed")
+
+    monkeypatch.setattr(registry, "_write_checkpoint", checkpoint)
+    with pytest.raises(RuntimeError):
+        registry._move_to_finished(session)
+    assert session._completion_event.is_set()
+
+
+def test_reader_exit_cannot_overwrite_a_committed_kill(monkeypatch):
+    """The reader records its exit under the session lock, so it sees a kill committed under it."""
+    registry = ProcessRegistry()
+    session = _session("proc_kill_vs_reader")
+    registry._running[session.id] = session
+    monkeypatch.setattr(registry, "_write_checkpoint", lambda *_a, **_k: None)
+
+    reader = threading.Thread(target=registry._finish_exited, args=(session, 0), daemon=True)
+    with session._lock:  # kill_process commits its result under this lock
+        reader.start()
+        time.sleep(0.2)
+        recorded_early = session.exited
+        session.exited = True
+        session.exit_code = -15
+        session.completion_reason = "killed"
+        session.termination_source = "process.kill"
+    reader.join(5)
+
+    assert not recorded_early, "reader recorded its exit without the session lock"
+    assert (session.exit_code, session.completion_reason, session.termination_source) == (
+        -15, "killed", "process.kill")
+
+
+def _finalizing_session(registry: ProcessRegistry) -> ProcessSession:
+    """Direct child exited and a reconcile flipped ``exited``; the reader is still on its final drain."""
+    session = _session("proc_finalizing", notify_on_complete=True)
+    session.process = cast(Any, SimpleNamespace(poll=lambda: 0, stdout=None, stderr=None, stdin=None))
+    session._reader_selectable = True
+    session._reader_thread = cast(Any, SimpleNamespace(is_alive=lambda: True))
+    session._reader_finish_requested.set()
+    session.mark_exited(0)
+    session.append_output("partial")
+    registry._running[session.id] = session
+    return session
+
+
+def test_status_reads_do_not_consume_a_completion_still_being_finalized():
+    registry = ProcessRegistry()
+    session = _finalizing_session(registry)
+
+    assert registry.poll(session.id)["status"] == "exited"
+    registry.read_log(session.id)
+
+    assert session.id not in registry._poll_observed
+    assert session.id not in registry._completion_consumed
+
+
+class _ReleaseHookLock:
+    """Session lock that runs a hook once, right after its first release."""
+
+    def __init__(self, hook):
+        self._lock, self._hook = threading.Lock(), hook
+
+    def acquire(self, *args, **kwargs):
+        return self._lock.acquire(*args, **kwargs)
+
+    __enter__ = acquire
+
+    def release(self):
+        self._lock.release()
+        hook, self._hook = self._hook, None
+        if hook:
+            hook()
+
+    def __exit__(self, *_exc):
+        self.release()
+
+
+@pytest.mark.parametrize("read", [
+    lambda registry, sid: registry.poll(sid),
+    lambda registry, sid: registry.read_log(sid),
+    lambda registry, sid: registry.kill_process(sid),
+], ids=["poll", "read_log", "kill_exited"])
+def test_a_snapshot_taken_before_the_reader_finishes_does_not_consume_it(read, monkeypatch):
+    """The reader finishes between the snapshot and the consume decision: the decision must
+    describe the snapshot returned, so the completion still delivers the whole output."""
+    registry = ProcessRegistry()
+    monkeypatch.setattr(registry, "_write_checkpoint", lambda *_a, **_k: None)
+    session = _finalizing_session(registry)
+
+    def reader_finishes():
+        session.append_output("FINAL")
+        registry._finish_exited(session, 0)
+
+    session._lock = cast(Any, _ReleaseHookLock(reader_finishes))
+    read(registry, session.id)
+
+    assert session._completion_event.is_set()
+    assert session.id not in registry._poll_observed
+    assert not registry.is_completion_consumed(session.id)
+    assert registry.completion_queue.get_nowait()["output"].endswith("FINAL")
+
+
+@pytest.mark.platforms("posix")
+def test_wait_returns_the_final_tail_when_the_reader_drains_slowly(tmp_path, monkeypatch):
+    """``wait`` must not take (and consume) the output before the reader has drained the exited
+    child's tail, however long the reader takes."""
+    monkeypatch.setattr(module, "_SYSTEMD_SCOPE_AVAILABLE", False, raising=False)
+    registry = ProcessRegistry()
+    stalled, release = threading.Event(), threading.Event()
+
+    def slow_sink(_session, _chunk):  # a live-output consumer that falls behind
+        if not stalled.is_set():
+            stalled.set()
+            release.wait(10)
+
+    registry.on_output = slow_sink
+    burst = ("import sys; sys.stdout.write('x' * 99 + '\\n'); "
+             "sys.stdout.write(('y' * 99 + '\\n') * 100 + 'FINAL\\n')")
+    session = registry.spawn_local(f"{shlex.quote(sys.executable)} -c {shlex.quote(burst)}", cwd=str(tmp_path))
+    session.notify_on_complete = True
+    try:
+        assert stalled.wait(10)
+        assert session.process.wait(timeout=10) == 0
+        # Longer than any bounded wait inside the reconcile.
+        threading.Timer(2.5, release.set).start()
+        result = registry.wait(session.id, timeout=15)
+        assert result["status"] == "exited"
+        assert result["output"].rstrip().endswith("FINAL"), result["output"][-80:]
+    finally:
+        release.set()
+        registry.kill_all(source="test")
+
+
+def test_kill_with_a_failed_scope_stop_keeps_the_session_running(monkeypatch):
+    """Without a stopped scope, nothing proves the detached descendants are gone."""
+    registry = ProcessRegistry()
+    session = _session("proc_scope_stuck", systemd_unit="hermes-worker-proc_scope_stuck.scope")
+    registry._running[session.id] = session
+    monkeypatch.setattr(registry, "_signal_kill", lambda *_args: None)
+    monkeypatch.setattr(registry, "_post_kill_survivors", lambda _session: [])
+    monkeypatch.setattr(registry, "_write_checkpoint", lambda *_a, **_k: None)
+    monkeypatch.setattr(module, "_stop_systemd_unit", lambda _unit: False)
+
+    result = registry.kill_process(session.id)
+
+    assert result["status"] == "error"
+    assert result["process_running"] is True
+    assert "hermes-worker-proc_scope_stuck.scope" in result["error"]
+    assert session.id in registry._running
+    assert not session.exited
+
+
+def test_kill_of_an_exited_session_reports_a_failed_scope_stop(monkeypatch):
+    registry = ProcessRegistry()
+    session = _session("proc_exited_scope_stuck", systemd_unit="hermes-worker-proc_exited_scope_stuck.scope")
+    session.mark_exited(0)
+    registry._finished[session.id] = session
+    monkeypatch.setattr(module, "_stop_systemd_unit", lambda _unit: False)
+
+    result = registry.kill_process(session.id)
+
+    assert result["status"] == "already_exited"
+    assert result["scope_stop_failed"] == "hermes-worker-proc_exited_scope_stuck.scope"
+
+
+def test_a_reader_exit_during_a_failed_scope_stop_is_not_reported_as_killed(monkeypatch):
+    """The root exiting proves nothing about the scope's detached descendants."""
+    registry = ProcessRegistry()
+    session = _session("proc_scope_race", systemd_unit="hermes-worker-proc_scope_race.scope")
+    registry._running[session.id] = session
+    monkeypatch.setattr(registry, "_signal_kill", lambda *_args: None)
+    monkeypatch.setattr(registry, "_write_checkpoint", lambda *_a, **_k: None)
+
+    def stop_while_the_reader_finishes(_unit):
+        registry._finish_exited(session, 0)
+        return False
+
+    monkeypatch.setattr(module, "_stop_systemd_unit", stop_while_the_reader_finishes)
+    result = registry.kill_process(session.id)
+
+    assert result["status"] == "error"
+    assert result["scope_stop_failed"] == "hermes-worker-proc_scope_race.scope"
+    assert result["process_running"] is False
+    assert (session.completion_reason, session.exit_code) == ("exited", 0)
+    assert session.id in registry._finished  # keeps its unit, so a later kill retries the stop

tools/AGENTS.md

@@ -108,8 +108,8 @@ _record_scope_trust` keys trust on the home; a secondary never adopts the launch
 for a same-named server, and `mcp_tool_handlers.py::_trust_gate_check` consults the calling
 session's profile.
 
-**Background-process teardown signals the parent first.** `process_registry.py::ProcessRegistry.
-_terminate_host_pid` snapshots the descendants, SIGTERMs only the recorded parent, waits
+**Background-process teardown signals the parent first.** `process_registry_termination.py::
+ProcessTerminationMixin._terminate_host_pid` snapshots the descendants, SIGTERMs only the recorded parent, waits
 `terminal.daemon_term_grace_seconds` for it to exit and reap its own children, then SIGTERMs the
 snapshot survivors and SIGKILLs whatever ignored both (so a supervisor that reaps its tree — a
 Chromium/Electron browser reaping its zygotes, a shell trap — exits cleanly, while a shell whose

tools/process_registry.py

@@ -34,6 +34,7 @@ from hermes_cli.config import get_hermes_home
 
 from tools.process_registry_notifications import format_process_notification
 from tools.process_registry_checkpoint import ProcessCheckpointMixin
+from tools.process_registry_termination import ProcessTerminationMixin
 from tools.process_registry_results import load_completed_results, save_completed_result
 
 logger = logging.getLogger(__name__)
@@ -592,6 +593,9 @@ class ProcessSession:
     _watch_strike_candidate: bool = field(default=False, repr=False)
     _watch_consecutive_strikes: int = field(default=0, repr=False)
     _completion_event: threading.Event = field(default_factory=threading.Event, repr=False)
+    # Set by the _move_to_finished call that moved the session out of _running; only that call
+    # publishes the completion, so a duplicate finisher must not release waiters early.
+    _finish_claimed: bool = field(default=False, repr=False)
     _lock: threading.Lock = field(default_factory=threading.Lock)
     _reader_thread: Optional[threading.Thread] = field(default=None, repr=False)
     _reader_finish_requested: threading.Event = field(default_factory=threading.Event, repr=False)
@@ -688,7 +692,7 @@ def _is_wsl_launcher_command(command: str) -> bool:
     return False
 
 
-class ProcessRegistry(ProcessCheckpointMixin):
+class ProcessRegistry(ProcessTerminationMixin, ProcessCheckpointMixin):
     """In-memory registry of running and finished background processes.
     Thread-safe: accessed from executor threads (terminal_tool, process handlers),
     the gateway asyncio loop (watchers, reset checks) and the cleanup thread."""
@@ -1070,15 +1074,6 @@ class ProcessRegistry(ProcessCheckpointMixin):
             and session.detached and session.pid_scope == "host"
             and not self._is_host_pid_alive(session.pid))
 
-    @staticmethod
-    def _proc_alive(proc) -> bool:
-        """True if a psutil.Process is running and not a zombie (already dead, just unreaped)."""
-        try:
-            import psutil
-            return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE
-        except Exception:
-            return False
-
     @staticmethod
     def _config_value(section: str, key: str, fallback):
         """``config.yaml`` value for ``section.key``, else the DEFAULT_CONFIG value.
@@ -1102,164 +1097,6 @@ class ProcessRegistry(ProcessCheckpointMixin):
         """Grace (s) between SIGTERM and escalated SIGKILL; 0 disables escalation."""
         return ProcessRegistry._config_seconds("daemon_term_grace_seconds", 2.0)
 
-    @classmethod
-    def _terminate_host_pid(cls, pid: int, expected_start: Optional[int] = None) -> None:
-        """Terminate a host-visible PID and its descendants.
-        ``expected_start`` (kernel start time at spawn) is re-validated first: a mismatch
-        or dead PID means the number was recycled onto a stranger and we refuse to touch
-        it — a leaked orphan beats tree-killing someone's browser. POSIX: snapshot descendants,
-        SIGTERM the parent alone so it can perform an orderly shutdown, then clean up snapshot
-        descendants that survive its grace window. Survivors are SIGKILLed after a second
-        ``terminal.daemon_term_grace_seconds`` window. Windows:
-        ``taskkill /T /F`` (psutil's stale PPID links miss orphans there); ``os.kill``
-        is the fallback."""
-        if expected_start is not None and not cls._host_pid_is_ours(pid, expected_start):
-            logger.warning(
-                "Refusing to terminate host pid %d: start-time mismatch — "
-                "PID was recycled onto an unrelated process.", pid)
-            return
-
-        def _sigterm_quietly():
-            with suppress(OSError, ProcessLookupError, PermissionError):
-                os.kill(pid, signal.SIGTERM)
-        if _IS_WINDOWS:
-            try:
-                subprocess.run(
-                    ["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True,
-                    encoding='utf-8', errors='replace', timeout=10, creationflags=windows_hide_flags(),
-                    stdin=subprocess.DEVNULL)
-            except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
-                _sigterm_quietly()
-            return
-        import psutil
-        gone = (psutil.NoSuchProcess, psutil.AccessDenied, OSError)
-        try:
-            parent = psutil.Process(pid)
-        except psutil.NoSuchProcess:
-            return
-        except (OSError, PermissionError):
-            _sigterm_quietly()
-            return
-        # Snapshot before signalling: once the parent exits, psutil can no longer
-        # reliably find children that it failed to reap.
-        try:
-            descendants = parent.children(recursive=True)
-        except gone:
-            descendants = []
-
-        # Let self-managing parents (notably Chromium/Electron) shut down their
-        # tree before touching children. Killing their zygotes first can turn a
-        # graceful browser shutdown into a crash dump.
-        with suppress(gone):
-            parent.terminate()
-
-        grace = cls._daemon_term_grace_seconds()
-
-        def _wait_for_exit(targets) -> None:
-            if grace <= 0:
-                return
-            deadline = time.monotonic() + grace
-            while time.monotonic() < deadline and any(cls._proc_alive(p) for p in targets):
-                time.sleep(0.05)
-
-        # Preserve descendants during the parent's configured shutdown window.
-        _wait_for_exit([parent])
-
-        # The snapshot is an anti-orphan guarantee: only descendants still alive
-        # after the parent had its chance are asked to terminate themselves.
-        remaining = descendants if grace <= 0 else [
-            proc for proc in descendants if cls._proc_alive(proc)
-        ]
-        for proc in remaining:
-            with suppress(gone):
-                proc.terminate()
-
-        # Preserve the existing SIGKILL escalation semantics for every owned
-        # process that remains after its SIGTERM grace window. The parent is
-        # included in case it ignored the first signal.
-        targets = [parent, *remaining]
-        # Escalate to SIGKILL for anything that ignored SIGTERM within the grace window.
-        # ``psutil.wait_procs``' gone/alive partition is deliberately NOT trusted: it
-        # reaps via ``Process.wait()`` and mis-partitions across zombie transitions in a
-        # parent/child tree, leaving survivors un-killed. Re-probing every target is
-        # deterministic.
-        if grace <= 0:
-            return
-        _wait_for_exit(targets)
-        # A parent that ignored SIGTERM (the interactive ``bash -lic`` wrapper does) keeps
-        # running its script through both grace windows and can spawn children the first
-        # snapshot never saw. Re-snapshot while it is still alive: once it is SIGKILLed
-        # they reparent to init and nothing can find them again.
-        with suppress(gone):
-            if cls._proc_alive(parent):
-                known = {proc.pid for proc in targets}
-                targets.extend(p for p in parent.children(recursive=True) if p.pid not in known)
-        for proc in targets:
-            with suppress(gone):
-                if cls._proc_alive(proc):
-                    proc.kill()  # SIGKILL on POSIX
-                    logger.info("Escalated to SIGKILL for pid %d (ignored SIGTERM within %.1fs grace)", proc.pid, grace)
-
-    @staticmethod
-    def _live_descendants(pid: int) -> List[int]:
-        """PIDs of living non-zombie descendants of host PID ``pid`` (best-effort)."""
-        try:
-            import psutil
-            children = psutil.Process(pid).children(recursive=True)
-        except Exception:
-            return []
-        return [c.pid for c in children if ProcessRegistry._proc_alive(c)]
-
-    # SIGKILL / taskkill are asynchronous: the kernel needs a scheduling tick to
-    # tear the process down and the parent must reap it before poll()/isalive()
-    # stop saying "alive". Verifying survivors in that window flagged every
-    # escalated kill as incomplete.
-    _KILL_SETTLE_SECONDS = 1.0
-
-    def _post_kill_survivors(self, session: "ProcessSession") -> List[int]:
-        """Host PIDs still alive once the kill signals have had time to land (#115490).
-
-        Fail-closed: anything unverifiable counts as a survivor, so a kill
-        that leaves a live tree can never write a killed receipt. Sandbox
-        (env) sessions have no host-visible tree and are unverifiable by
-        design — they return no survivors, preserving existing behavior."""
-        deadline = time.monotonic() + self._KILL_SETTLE_SECONDS
-        while True:
-            survivors = self._probe_survivors(session)
-            if not survivors or time.monotonic() >= deadline:
-                return survivors
-            time.sleep(0.05)
-
-    def _probe_survivors(self, session: "ProcessSession") -> List[int]:
-        survivors: List[int] = []
-        proc = getattr(session, "process", None)
-        if proc is not None:
-            try:
-                root_alive = proc.poll() is None
-            except Exception:
-                root_alive = True
-            if root_alive:
-                survivors.append(getattr(proc, "pid", None) or session.pid)
-        pty = getattr(session, "_pty", None)
-        if pty is not None:
-            try:
-                pty_alive = bool(pty.isalive())
-            except Exception:
-                pty_alive = self._is_host_pid_alive(session.pid)
-            if pty_alive:
-                survivors.append(session.pid)
-        if session.pid_scope == "host" and session.pid:
-            if self._detached_host_fate(session.pid, session.host_start_time) == "running":
-                if session.pid not in survivors:
-                    survivors.append(session.pid)
-                survivors.extend(
-                    pid for pid in self._live_descendants(session.pid)
-                    if pid not in survivors)
-            # A dead/recycled root has no PID-scope descendants left to find:
-            # reparented orphans are outside PID scope (systemd scope stop,
-            # issued before this check, covers the cgroup case).
-        return [pid for pid in survivors if pid]
-
     # ----- Spawn -----
 
     @staticmethod
@@ -1768,8 +1605,11 @@ class ProcessRegistry(ProcessCheckpointMixin):
         self._emit_output(session, text)
 
     def _finish_exited(self, session: ProcessSession, exit_code) -> None:
-        """Mark a reader-observed exit (a raced kill keeps its own code/reason) and finish."""
-        session.mark_exited(exit_code)
+        """Mark a reader-observed exit (a raced kill keeps its own code/reason) and finish.
+        Under the session lock, like kill_process's commit: otherwise the kill can land between
+        mark_exited's ``killed`` check and its writes, and the reader overwrites the kill."""
+        with session._lock:
+            session.mark_exited(exit_code)
         self._move_to_finished(session)
 
     def _move_to_finished(self, session: ProcessSession) -> bool:
@@ -1780,6 +1620,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
         with self._lock:
             was_running = session.id in self._running
             if was_running:
+                session._finish_claimed = True
                 session.exited_at = time.time()
                 # Keep the session tracked until its result is durable. A finite
                 # parent must not observe completion and exit during this write.
@@ -1795,6 +1636,16 @@ class ProcessRegistry(ProcessCheckpointMixin):
         # and the loop exits, dropping at most the unread tail of a process
         # that was just killed. poll()/wait()/read_log() serve from the
         # buffered ``output_buffer``, never from the pipe.
+        try:
+            self._publish_finished(session, was_running)
+        finally:
+            # The owner releases waiters only after its notice is queued (or its publish
+            # failed, so a waiter never parks forever). A duplicate leaves the event to it.
+            if was_running or not session._finish_claimed:
+                session._completion_event.set()
+        return was_running
+
+    def _publish_finished(self, session: ProcessSession, was_running: bool) -> None:
         self._release_finished_handles(session)
         self._write_checkpoint()
         if was_running and session.notify_on_complete:
@@ -1815,8 +1666,6 @@ class ProcessRegistry(ProcessCheckpointMixin):
             }
             _redact_process_result(notification)
             self.completion_queue.put(notification)
-        session._completion_event.set()
-        return was_running
 
     @staticmethod
     def _exit_fields(session: ProcessSession) -> dict:
@@ -1861,6 +1710,15 @@ class ProcessRegistry(ProcessCheckpointMixin):
 
     # ----- Query Methods -----
 
+    @staticmethod
+    def _reader_finalizing(session: ProcessSession) -> bool:
+        """A reconcile saw the direct child exit and set ``exited``, but the reader that owns the
+        output is still on its final drain: the buffer is not the whole output yet, so nothing
+        may consume it as the completion."""
+        reader = session._reader_thread
+        return (session._reader_finish_requested.is_set() and not session._completion_event.is_set()
+                and reader is not None and reader.is_alive())
+
     def is_completion_consumed(self, session_id: str) -> bool:
         """Check if a completion notification was already consumed via wait/log."""
         return session_id in self._completion_consumed
@@ -2200,17 +2058,22 @@ class ProcessRegistry(ProcessCheckpointMixin):
         if session is None or self._uncollected_gone(session):
             return _not_found(session_id)
         self._reconcile_local_exit(session)  # orphaned-pipe reader guard
+        # Read with the preview under the lock the reader appends and exits under, so the
+        # finalization check describes the output actually returned.
         with session._lock:
             output_preview = _output_tail(session, 1000)
+            exited, finalizing = session.exited, self._reader_finalizing(session)
         result = {
             **self._status_head(session), "pid": session.pid,
             "uptime_seconds": int(time.time() - session.started_at), "output_preview": output_preview}
-        if session.exited:
+        if exited:
             result.update(self._exit_fields(session))
             # Read-only: record in _poll_observed (CLI inline dedup) but NOT in
             # _completion_consumed, or a status check would suppress the watcher's
-            # autonomous delivery turn. See __init__.
-            self._poll_observed.add(session_id)
+            # autonomous delivery turn. See __init__. A preview taken before the reader's
+            # final drain is partial and must not suppress the full completion.
+            if not finalizing:
+                self._poll_observed.add(session_id)
         if session.detached:
             result.update(detached=True, note="Process recovered after restart -- output history unavailable")
         return result
@@ -2224,6 +2087,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
             return _not_found(session_id)
         with session._lock:
             full_output = strip_ansi(session.output_buffer)
+            exited, finalizing = session.exited, self._reader_finalizing(session)
         lines = full_output.splitlines()
         total_lines = len(lines)
         # offset=None -> last N lines; an explicit offset=0 means the HEAD (don't
@@ -2242,7 +2106,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
         result = {
             **self._status_head(session), "output": "\n".join(selected),
             "total_lines": total_lines, "showing": f"{len(selected)} lines"}
-        if session.exited and observed_completion_output:
+        if exited and observed_completion_output and not finalizing:
             self._completion_consumed.add(session_id)
         return result
 
@@ -2276,7 +2140,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
                 return _not_found(session_id)
             self._reconcile_local_exit(session)  # orphaned-pipe reader guard
             result = None
-            if session.exited:
+            if session.exited and not self._reader_finalizing(session):
                 self._completion_consumed.add(session_id)
                 result = self._exit_snapshot(session, "exited")
             elif _is_interrupted():
@@ -2344,13 +2208,15 @@ class ProcessRegistry(ProcessCheckpointMixin):
             # above (reviewer gap #2). ``systemctl --user stop`` sends SIGTERM to every process in the
             # cgroup and escalates to SIGKILL after TimeoutStopSec. This is additive — the PID-based kill
             # above already handled the main process; this catches stragglers.
-            if session.systemd_unit:
-                _stop_systemd_unit(session.systemd_unit)
+            scope_stopped = not session.systemd_unit or _stop_systemd_unit(session.systemd_unit)
             with session._lock:
                 result = self._exit_snapshot(session, "already_exited")
+                finalizing = self._reader_finalizing(session)
+            if not scope_stopped:
+                result["scope_stop_failed"] = session.systemd_unit
             # Only suppress the autonomous turn after its output is present in
             # the explicit kill result, matching wait/log consumption.
-            if consume_output:
+            if consume_output and not finalizing:
                 self._completion_consumed.add(session_id)
             return result
         try:
@@ -2359,8 +2225,7 @@ class ProcessRegistry(ProcessCheckpointMixin):
                 return early
             # Additive to the PID kill: stopping the scope reaps double-forked
             # descendants reparented inside the cgroup.
-            if session.systemd_unit:
-                _stop_systemd_unit(session.systemd_unit)
+            scope_stopped = not session.systemd_unit or _stop_systemd_unit(session.systemd_unit)
             # Post-kill verification (#115490): the signals above can leave
             # survivors (SIGTERM-ignoring daemons, scope escapees). A kill that
             # leaves a live tree must not write a killed receipt or prune the
@@ -2383,6 +2248,19 @@ class ProcessRegistry(ProcessCheckpointMixin):
                         f"({alive}); session running"),
                     "session_id": session.id, "survivors": survivors,
                     "process_running": True}
+            # Descendants outside the root's tree are visible only to the scope; a scope that did
+            # not stop leaves nothing proving they are gone, even when the root exited meanwhile.
+            # The session keeps its unit, so a later kill retries the stop.
+            if not scope_stopped:
+                with session._lock:
+                    root_exited = session.exited
+                logger.warning("Kill incomplete for %s: systemd scope %s did not stop",
+                               session.id, session.systemd_unit)
+                return {
+                    "status": "error",
+                    "error": f"Kill incomplete: systemd scope {session.systemd_unit} did not stop",
+                    "session_id": session.id, "scope_stop_failed": session.systemd_unit,
+                    "process_running": not root_exited}
             # Capture output, mark consumed, THEN expose ``exited`` to watcher tasks —
             # closes the delayed-notification race without losing the transcript.
             with session._lock:

tools/process_registry_termination.py

@@ -0,0 +1,191 @@
+"""Process-tree termination and post-kill survivor verification for ProcessRegistry."""
+
+import logging
+import os
+import signal
+import subprocess
+import time
+from contextlib import suppress
+from typing import TYPE_CHECKING, List, Optional
+
+from hermes_cli._subprocess_compat import windows_hide_flags
+
+if TYPE_CHECKING:
+    from tools.process_registry import ProcessSession
+
+logger = logging.getLogger("tools.process_registry")
+
+
+class ProcessTerminationMixin:
+    """The subclass supplies ``_host_pid_is_ours``, ``_is_host_pid_alive``,
+    ``_detached_host_fate`` and ``_daemon_term_grace_seconds``. See ProcessRegistry."""
+
+    @staticmethod
+    def _proc_alive(proc) -> bool:
+        """True if a psutil.Process is running and not a zombie (already dead, just unreaped)."""
+        try:
+            import psutil
+            return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE
+        except Exception:
+            return False
+
+    @classmethod
+    def _terminate_host_pid(cls, pid: int, expected_start: Optional[int] = None) -> None:
+        """Terminate a host-visible PID and its descendants.
+        ``expected_start`` (kernel start time at spawn) is re-validated first: a mismatch
+        or dead PID means the number was recycled onto a stranger and we refuse to touch
+        it — a leaked orphan beats tree-killing someone's browser. POSIX: snapshot descendants,
+        SIGTERM the parent alone so it can perform an orderly shutdown, then clean up snapshot
+        descendants that survive its grace window. Survivors are SIGKILLed after a second
+        ``terminal.daemon_term_grace_seconds`` window. Windows:
+        ``taskkill /T /F`` (psutil's stale PPID links miss orphans there); ``os.kill``
+        is the fallback."""
+        from tools.process_registry import _IS_WINDOWS
+
+        if expected_start is not None and not cls._host_pid_is_ours(pid, expected_start):
+            logger.warning(
+                "Refusing to terminate host pid %d: start-time mismatch — "
+                "PID was recycled onto an unrelated process.", pid)
+            return
+
+        def _sigterm_quietly():
+            with suppress(OSError, ProcessLookupError, PermissionError):
+                os.kill(pid, signal.SIGTERM)
+        if _IS_WINDOWS:
+            try:
+                subprocess.run(
+                    ["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True,
+                    encoding='utf-8', errors='replace', timeout=10, creationflags=windows_hide_flags(),
+                    stdin=subprocess.DEVNULL)
+            except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
+                _sigterm_quietly()
+            return
+        import psutil
+        gone = (psutil.NoSuchProcess, psutil.AccessDenied, OSError)
+        try:
+            parent = psutil.Process(pid)
+        except psutil.NoSuchProcess:
+            return
+        except (OSError, PermissionError):
+            _sigterm_quietly()
+            return
+        # Snapshot before signalling: once the parent exits, psutil can no longer
+        # reliably find children that it failed to reap.
+        try:
+            descendants = parent.children(recursive=True)
+        except gone:
+            descendants = []
+
+        # Let self-managing parents (notably Chromium/Electron) shut down their
+        # tree before touching children. Killing their zygotes first can turn a
+        # graceful browser shutdown into a crash dump.
+        with suppress(gone):
+            parent.terminate()
+
+        grace = cls._daemon_term_grace_seconds()
+
+        def _wait_for_exit(targets) -> None:
+            if grace <= 0:
+                return
+            deadline = time.monotonic() + grace
+            while time.monotonic() < deadline and any(cls._proc_alive(p) for p in targets):
+                time.sleep(0.05)
+
+        # Preserve descendants during the parent's configured shutdown window.
+        _wait_for_exit([parent])
+
+        # The snapshot is an anti-orphan guarantee: only descendants still alive
+        # after the parent had its chance are asked to terminate themselves.
+        remaining = descendants if grace <= 0 else [
+            proc for proc in descendants if cls._proc_alive(proc)
+        ]
+        for proc in remaining:
+            with suppress(gone):
+                proc.terminate()
+
+        # Preserve the existing SIGKILL escalation semantics for every owned
+        # process that remains after its SIGTERM grace window. The parent is
+        # included in case it ignored the first signal.
+        targets = [parent, *remaining]
+        # Escalate to SIGKILL for anything that ignored SIGTERM within the grace window.
+        # ``psutil.wait_procs``' gone/alive partition is deliberately NOT trusted: it
+        # reaps via ``Process.wait()`` and mis-partitions across zombie transitions in a
+        # parent/child tree, leaving survivors un-killed. Re-probing every target is
+        # deterministic.
+        if grace <= 0:
+            return
+        _wait_for_exit(targets)
+        # A parent that ignored SIGTERM (the interactive ``bash -lic`` wrapper does) keeps
+        # running its script through both grace windows and can spawn children the first
+        # snapshot never saw. Re-snapshot while it is still alive: once it is SIGKILLed
+        # they reparent to init and nothing can find them again.
+        with suppress(gone):
+            if cls._proc_alive(parent):
+                known = {proc.pid for proc in targets}
+                targets.extend(p for p in parent.children(recursive=True) if p.pid not in known)
+        for proc in targets:
+            with suppress(gone):
+                if cls._proc_alive(proc):
+                    proc.kill()  # SIGKILL on POSIX
+                    logger.info("Escalated to SIGKILL for pid %d (ignored SIGTERM within %.1fs grace)", proc.pid, grace)
+
+    @staticmethod
+    def _live_descendants(pid: int) -> List[int]:
+        """PIDs of living non-zombie descendants of host PID ``pid`` (best-effort)."""
+        try:
+            import psutil
+            children = psutil.Process(pid).children(recursive=True)
+        except Exception:
+            logger.debug("Could not list descendants of pid %s", pid, exc_info=True)
+            return []
+        return [c.pid for c in children if ProcessTerminationMixin._proc_alive(c)]
+
+    # SIGKILL / taskkill are asynchronous: the kernel needs a scheduling tick to
+    # tear the process down and the parent must reap it before poll()/isalive()
+    # stop saying "alive". Verifying survivors in that window flagged every
+    # escalated kill as incomplete.
+    _KILL_SETTLE_SECONDS = 1.0
+
+    def _post_kill_survivors(self, session: "ProcessSession") -> List[int]:
+        """Host PIDs still alive once the kill signals have had time to land (#115490).
+
+        Fail-closed: anything unverifiable counts as a survivor, so a kill
+        that leaves a live tree can never write a killed receipt. Sandbox
+        (env) sessions have no host-visible tree and are unverifiable by
+        design — they return no survivors, preserving existing behavior."""
+        deadline = time.monotonic() + self._KILL_SETTLE_SECONDS
+        while True:
+            survivors = self._probe_survivors(session)
+            if not survivors or time.monotonic() >= deadline:
+                return survivors
+            time.sleep(0.05)
+
+    def _probe_survivors(self, session: "ProcessSession") -> List[int]:
+        survivors: List[int] = []
+        proc = getattr(session, "process", None)
+        if proc is not None:
+            try:
+                root_alive = proc.poll() is None
+            except Exception:
+                root_alive = True
+            if root_alive:
+                survivors.append(getattr(proc, "pid", None) or session.pid)
+        pty = getattr(session, "_pty", None)
+        if pty is not None:
+            try:
+                pty_alive = bool(pty.isalive())
+            except Exception:
+                pty_alive = self._is_host_pid_alive(session.pid)
+            if pty_alive:
+                survivors.append(session.pid)
+        if session.pid_scope == "host" and session.pid:
+            if self._detached_host_fate(session.pid, session.host_start_time) == "running":
+                if session.pid not in survivors:
+                    survivors.append(session.pid)
+                survivors.extend(
+                    pid for pid in self._live_descendants(session.pid)
+                    if pid not in survivors)
+            # A dead/recycled root has no PID-scope descendants left to find:
+            # reparented orphans are outside PID scope (systemd scope stop,
+            # issued before this check, covers the cgroup case).
+        return [pid for pid in survivors if pid]