diff --git a/tests/tools/test_process_registry_completion_races.py b/tests/tools/test_process_registry_completion_races.py
new file mode 100644
index 0000000000..eb5b667a42
--- /dev/null
+++ b/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
diff --git a/tools/AGENTS.md b/tools/AGENTS.md
index e6b302e951..d5c8290f20 100644
--- a/tools/AGENTS.md
+++ b/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
diff --git a/tools/process_registry.py b/tools/process_registry.py
index da31d829ec..256279e14f 100644
--- a/tools/process_registry.py
+++ b/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:
diff --git a/tools/process_registry_termination.py b/tools/process_registry_termination.py
new file mode 100644
index 0000000000..c917b1fa7d
--- /dev/null
+++ b/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]