From 6cd9ac24992de4f26ca97f1217e9d487dcd99c95 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 19 Oct 2026 03:57:27 +0000
Subject: [PATCH 01/12] chore: record implementation plan


From cc2741970da0a00aa3944bd1ce117c73cc3c7411 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 19 Oct 2026 04:00:15 +0000
Subject: [PATCH 02/12] fix(evals): manage isolated daemon lifecycle

Co-authored-by: ScriptedAlchemy <25274700+ScriptedAlchemy@users.noreply.github.com>
---
 evals/memory/README.md              |   5 +
 evals/memory/run_real_model.py      | 264 ++++++++++++++++++++++------
 evals/memory/test_run_real_model.py | 112 ++++++++++++
 3 files changed, 332 insertions(+), 49 deletions(-)

diff --git a/evals/memory/README.md b/evals/memory/README.md
index cc8d813f44..8dcf129adb 100644
--- a/evals/memory/README.md
+++ b/evals/memory/README.md
@@ -30,6 +30,11 @@ python3 evals/memory/run_real_model.py --scenario memory-no-pollution \
 The real-model runner creates a unique temporary user home for each run and
 installs the normal user-level Hermes integration there. It never selects a
 named Hermes profile or uses `HERMES_HOME` as TraceDecay storage.
+Each scenario also owns an isolated `tracedecay daemon run` (private socket,
+started via `scripts/lib/portable_process.py`) that must be ready before
+`init`. Setup, daemon, agent, and cancellation failures are written to the
+report as error entries, and the daemon, fixture, and profile are removed on
+every exit path.
 
 Run reports land under `evals/memory/runs/<timestamp>/` (gitignored).
 
diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index b4aa1d363e..3aab8f7c0d 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -29,6 +29,7 @@
 import subprocess
 import sys
 import tempfile
+import signal
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -36,6 +37,9 @@
 REPO_ROOT = EVAL_DIR.parent.parent
 SCENARIO_DIR = EVAL_DIR / "scenarios"
 RUNS_DIR = EVAL_DIR / "runs"
+PROCESS_HELPER = REPO_ROOT / "scripts/lib/portable_process.py"
+DAEMON_READY_TIMEOUT_SECONDS = 60
+DAEMON_STOP_GRACE_SECONDS = 5
 
 DEFAULT_HERMES_DIR = Path.home() / "hermes-agent"
 DEFAULT_MODEL = "gpt-5.4-mini"
@@ -120,15 +124,110 @@ class EvalEnvironment:
     global_db: Path
     env: dict[str, str]
     _temp_dir: tempfile.TemporaryDirectory
+    socket_dir: Path | None = None
+    daemon: subprocess.Popen | None = None
+    daemon_log: Path | None = None
+
+    def start_daemon(self, tracedecay_bin, ready_timeout=DAEMON_READY_TIMEOUT_SECONDS):
+        """Launches the selected binary's daemon on a socket private to this profile."""
+        if self.daemon is not None:
+            raise EvalSetupError("isolated daemon already started")
+        # Unix socket paths are length-limited, so keep the socket in a short directory.
+        self.socket_dir = Path(tempfile.mkdtemp(prefix="td-eval-"))
+        socket_path = self.socket_dir / "daemon.sock"
+        self.env["TRACEDECAY_DAEMON_SOCKET"] = str(socket_path)
+        self.daemon_log = self.root / "daemon.log"
+        with open(self.daemon_log, "wb") as log:
+            self.daemon = subprocess.Popen(
+                [
+                    sys.executable,
+                    "-S",
+                    str(PROCESS_HELPER),
+                    "exec-session",
+                    "--parent-pid",
+                    str(os.getpid()),
+                    "--",
+                    tracedecay_bin,
+                    "daemon",
+                    "run",
+                    "--socket",
+                    str(socket_path),
+                ],
+                env=self.env,
+                stdout=log,
+                stderr=subprocess.STDOUT,
+            )
+        ready = subprocess.run(
+            [
+                sys.executable,
+                "-S",
+                str(PROCESS_HELPER),
+                "wait-unix-socket",
+                "--path",
+                str(socket_path),
+                "--pid",
+                str(self.daemon.pid),
+                "--timeout",
+                str(ready_timeout),
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+        if ready.returncode != 0:
+            raise EvalSetupError(
+                f"isolated daemon did not become ready ({ready.returncode}): "
+                f"{ready.stdout.strip()}\n{self.daemon_log_tail()}"
+            )
+
+    def daemon_log_tail(self):
+        if self.daemon_log is None or not self.daemon_log.exists():
+            return "(no daemon output captured)"
+        return self.daemon_log.read_text(errors="replace")[-4096:] or "(no daemon output captured)"
+
+    def stop_daemon(self):
+        daemon, self.daemon = self.daemon, None
+        if daemon is None:
+            return
+        subprocess.run(
+            [
+                sys.executable,
+                "-S",
+                str(PROCESS_HELPER),
+                "stop-group",
+                "--pid",
+                str(daemon.pid),
+                "--grace",
+                str(DAEMON_STOP_GRACE_SECONDS),
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        try:
+            daemon.wait(timeout=DAEMON_STOP_GRACE_SECONDS)
+        except subprocess.TimeoutExpired:
+            daemon.kill()
+            daemon.wait()
 
     def cleanup(self):
-        self._temp_dir.cleanup()
+        try:
+            self.stop_daemon()
+        finally:
+            if self.socket_dir is not None:
+                shutil.rmtree(self.socket_dir, ignore_errors=True)
+            self._temp_dir.cleanup()
+
+
+class EvalSetupError(RuntimeError):
+    """An operational failure (daemon, init, install) reported instead of exiting."""
 
 
 def cleanup_eval_artifacts(fixture, eval_env):
-    if fixture is not None:
-        shutil.rmtree(fixture, ignore_errors=True)
-    eval_env.cleanup()
+    try:
+        if fixture is not None:
+            shutil.rmtree(fixture, ignore_errors=True)
+    finally:
+        eval_env.cleanup()
 
 
 def create_eval_environment(scenario_id):
@@ -203,7 +302,9 @@ def run(cmd, cwd=None, env=None, timeout=None, check=True):
         timeout=timeout,
     )
     if check and result.returncode != 0:
-        sys.exit(f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}")
+        raise EvalSetupError(
+            f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}"
+        )
     return result
 
 
@@ -242,8 +343,24 @@ def require_fact_record(response, operation):
     return fact
 
 
-def build_fixture(scenario, tracedecay_bin, env):
-    fixture = Path(tempfile.mkdtemp(prefix=f"tracedecay-eval-{scenario['id']}-"))
+def build_fixture(scenario, tracedecay_bin, env, parent=None):
+    """Creates and seeds the fixture; removes it again if any setup step fails."""
+    safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario["id"])
+    fixture = Path(
+        tempfile.mkdtemp(
+            prefix=f"tracedecay-eval-{safe_id}-",
+            dir=str(parent) if parent is not None else None,
+        )
+    )
+    try:
+        seed_fixture(scenario, tracedecay_bin, env, fixture)
+    except BaseException:
+        shutil.rmtree(fixture, ignore_errors=True)
+        raise
+    return fixture
+
+
+def seed_fixture(scenario, tracedecay_bin, env, fixture):
     (fixture / "src").mkdir()
     (fixture / "src/lib.rs").write_text("pub fn eval_fixture_marker() {}\n")
     for name, contents in scenario["setup"].get("files", {}).items():
@@ -287,7 +404,6 @@ def build_fixture(scenario, tracedecay_bin, env):
                     f"fixture preload search `{preload_query}` did not return "
                     f"FactId {seeded_fact['fact_id']}"
                 )
-    return fixture
 
 
 def provider_env_passthrough(env):
@@ -685,6 +801,77 @@ def evaluate_assertions(scenario, tracedecay_bin, fixture, env):
     return outcomes
 
 
+def run_scenario(args, scenario, run_dir, report):
+    """Runs one scenario in an owned profile/daemon/fixture scope.
+
+    Returns (ok, cancelled). Operational failures and cancellation are recorded
+    in the report; the daemon, fixture, and profile are removed on every path.
+    """
+    eval_env = create_eval_environment(scenario["id"])
+    fixture = None
+    try:
+        eval_env.start_daemon(args.tracedecay_bin)
+        fixture = build_fixture(
+            scenario, args.tracedecay_bin, eval_env.env, parent=eval_env.root
+        )
+        if args.driver == "hermes":
+            transcripts = drive_hermes(args, scenario, fixture, run_dir, eval_env.env)
+        else:
+            transcripts = drive_cursor_agent(args, scenario, fixture, run_dir, eval_env.env)
+        outcomes = evaluate_assertions(scenario, args.tracedecay_bin, fixture, eval_env.env)
+    except KeyboardInterrupt as exc:
+        report["scenarios"].append(error_entry(scenario, "cancelled", str(exc) or "interrupted"))
+        print(f"[{scenario['id']}] cancelled")
+        return False, True
+    except Exception as exc:
+        report["scenarios"].append(
+            error_entry(scenario, "error (setup or agent failure)", str(exc))
+        )
+        print(f"[{scenario['id']}] error: {exc}", file=sys.stderr)
+        return False, False
+    finally:
+        cleanup_eval_artifacts(fixture, eval_env)
+
+    failed = [o for o in outcomes if not o["passed"]]
+    status = "pass" if not failed else "fail"
+    if failed and scenario.get("contract") == "pending-sibling":
+        status = "fail (note: scenario contract is pending-sibling, see contract_notes)"
+    if not all(t["turn_valid"] for t in transcripts):
+        status = "error (agent turn invalid, see transcript logs)"
+        failed = failed or [{"name": "agent-turn", "passed": False}]
+    report["scenarios"].append(
+        {
+            "id": scenario["id"],
+            "contract": scenario.get("contract", "stable"),
+            "status": status,
+            "assertions": outcomes,
+            "transcripts": transcripts,
+        }
+    )
+    print(f"[{scenario['id']}] {status}")
+    for outcome in outcomes:
+        marker = "pass" if outcome["passed"] else "FAIL"
+        if outcome.get("error"):
+            print(f"  [{marker}] {outcome['name']}, {outcome['error']}")
+        else:
+            print(
+                f"  [{marker}] {outcome['name']}, actual {outcome['actual']} "
+                f"{outcome['op']} expected {outcome['expected']}"
+            )
+    return not failed, False
+
+
+def error_entry(scenario, status, error):
+    return {
+        "id": scenario["id"],
+        "contract": scenario.get("contract", "stable"),
+        "status": status,
+        "error": error,
+        "assertions": [],
+        "transcripts": [],
+    }
+
+
 def main(argv):
     args = parse_args(argv)
     args.tracedecay_bin = resolve_tracedecay_bin(args.tracedecay_bin)
@@ -717,48 +904,27 @@ def main(argv):
         print(f"\nblocked report written to {report_path}", file=sys.stderr)
         return 2
 
+    def raise_cancelled(signum, _frame):
+        raise KeyboardInterrupt(f"cancelled by signal {signum}")
+
+    previous_term = signal.signal(signal.SIGTERM, raise_cancelled)
     overall_ok = True
-    for scenario in scenarios:
-        eval_env = create_eval_environment(scenario["id"])
-        fixture = None
-        try:
-            fixture = build_fixture(scenario, args.tracedecay_bin, eval_env.env)
-            if args.driver == "hermes":
-                transcripts = drive_hermes(args, scenario, fixture, run_dir, eval_env.env)
-            else:
-                transcripts = drive_cursor_agent(args, scenario, fixture, run_dir, eval_env.env)
-            outcomes = evaluate_assertions(
-                scenario, args.tracedecay_bin, fixture, eval_env.env
-            )
-            failed = [o for o in outcomes if not o["passed"]]
-            status = "pass" if not failed else "fail"
-            if failed and scenario.get("contract") == "pending-sibling":
-                status = "fail (note: scenario contract is pending-sibling, see contract_notes)"
-            if not all(t["turn_valid"] for t in transcripts):
-                status = "error (agent turn invalid, see transcript logs)"
-                failed = failed or [{"name": "agent-turn", "passed": False}]
-            report["scenarios"].append(
-                {
-                    "id": scenario["id"],
-                    "contract": scenario.get("contract", "stable"),
-                    "status": status,
-                    "assertions": outcomes,
-                    "transcripts": transcripts,
-                }
-            )
-            overall_ok &= not failed
-            print(f"[{scenario['id']}] {status}")
-            for outcome in outcomes:
-                marker = "pass" if outcome["passed"] else "FAIL"
-                if outcome.get("error"):
-                    print(f"  [{marker}] {outcome['name']}, {outcome['error']}")
-                else:
-                    print(
-                        f"  [{marker}] {outcome['name']}, actual {outcome['actual']} "
-                        f"{outcome['op']} expected {outcome['expected']}"
-                    )
-        finally:
-            cleanup_eval_artifacts(fixture, eval_env)
+    cancelled = False
+    try:
+        for scenario in scenarios:
+            ok, cancelled = run_scenario(args, scenario, run_dir, report)
+            overall_ok &= ok
+            if cancelled:
+                break
+    finally:
+        signal.signal(signal.SIGTERM, previous_term)
+
+    if cancelled:
+        report["status"] = "cancelled"
+        report_path = run_dir / "report.json"
+        report_path.write_text(json.dumps(report, indent=2) + "\n")
+        print(f"\ncancelled; report written to {report_path}", file=sys.stderr)
+        return 130
 
     report["status"] = "pass" if overall_ok else "fail"
     report_path = run_dir / "report.json"
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index 03d7a87d07..57cb07312f 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -3,6 +3,7 @@
 import importlib.util
 import io
 import json
+import os
 import subprocess
 import tempfile
 import unittest
@@ -125,6 +126,117 @@ def test_cleanup_removes_fixture_and_isolated_eval_store(self):
             self.assertFalse(eval_env.root.exists())
 
 
+FAKE_BIN = """#!/usr/bin/env python3
+import os, socket, sys, time
+args = sys.argv[1:]
+if args[:2] == ["daemon", "run"]:
+    if os.environ.get("FAKE_DAEMON_FAIL"):
+        sys.exit(3)
+    path = args[3]
+    srv = socket.socket(socket.AF_UNIX)
+    srv.bind(path)
+    srv.listen(8)
+    open(os.environ["TRACEDECAY_DATA_DIR"] + "/daemon.pid", "w").write(str(os.getpid()))
+    time.sleep(600)
+elif args[:1] == ["init"]:
+    if os.environ.get("FAKE_INIT_FAIL"):
+        print("code_index_scheduler_unavailable")
+        sys.exit(1)
+    c = socket.socket(socket.AF_UNIX)
+    c.connect(os.environ["TRACEDECAY_DAEMON_SOCKET"])
+elif args[:1] == ["tool"]:
+    print('{"fact": {"fact_id": "fact.1"}}')
+"""
+
+
+class IsolatedLifecycleTest(unittest.TestCase):
+    def make_bin(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        path = Path(tmp.name) / "tracedecay"
+        path.write_text(FAKE_BIN)
+        path.chmod(0o755)
+        return str(path)
+
+    def scenario(self):
+        return {
+            "id": "lifecycle",
+            "setup": {
+                "facts": [
+                    {"content": "c", "category": "project", "source": "s", "trust": 0.5}
+                ]
+            },
+        }
+
+    def test_fresh_environment_inits_and_seeds_against_isolated_daemon(self):
+        binary = self.make_bin()
+        eval_env = run_real_model.create_eval_environment("lifecycle")
+        try:
+            eval_env.start_daemon(binary, ready_timeout=15)
+            fixture = run_real_model.build_fixture(
+                self.scenario(), binary, eval_env.env, parent=eval_env.root
+            )
+            self.assertTrue(fixture.is_relative_to(eval_env.root))
+            pid = int((eval_env.data_dir / "daemon.pid").read_text())
+        finally:
+            run_real_model.cleanup_eval_artifacts(None, eval_env)
+        self.assertFalse(eval_env.root.exists())
+        self.assertFalse(Path(eval_env.env["TRACEDECAY_DAEMON_SOCKET"]).parent.exists())
+        with self.assertRaises(ProcessLookupError):
+            os.kill(pid, 0)
+
+    def test_init_failure_removes_fixture_and_raises_typed_error(self):
+        binary = self.make_bin()
+        eval_env = run_real_model.create_eval_environment("lifecycle")
+        self.addCleanup(eval_env.cleanup)
+        eval_env.start_daemon(binary, ready_timeout=15)
+        eval_env.env["FAKE_INIT_FAIL"] = "1"
+        with self.assertRaises(run_real_model.EvalSetupError):
+            run_real_model.build_fixture(
+                self.scenario(), binary, eval_env.env, parent=eval_env.root
+            )
+        self.assertEqual(
+            [p.name for p in eval_env.root.glob("tracedecay-eval-*")], []
+        )
+
+    def test_daemon_startup_failure_is_reported_and_cleaned(self):
+        binary = self.make_bin()
+        eval_env = run_real_model.create_eval_environment("lifecycle")
+        eval_env.env["FAKE_DAEMON_FAIL"] = "1"
+        try:
+            with self.assertRaises(run_real_model.EvalSetupError):
+                eval_env.start_daemon(binary, ready_timeout=15)
+        finally:
+            eval_env.cleanup()
+        self.assertFalse(eval_env.root.exists())
+
+    def run_scenario_with(self, binary, driver):
+        args = argparse.Namespace(tracedecay_bin=binary, driver="hermes")
+        report = {"scenarios": []}
+        with tempfile.TemporaryDirectory() as run_dir, mock.patch.object(
+            run_real_model, "drive_hermes", side_effect=driver
+        ):
+            result = run_real_model.run_scenario(
+                args, self.scenario(), Path(run_dir), report
+            )
+        return result, report
+
+    def test_agent_failure_and_cancellation_write_error_entries(self):
+        binary = self.make_bin()
+        for exc, status, cancelled in (
+            (RuntimeError("agent exploded"), "error", False),
+            (KeyboardInterrupt("stop"), "cancelled", True),
+        ):
+            with self.subTest(status=status):
+                result, report = self.run_scenario_with(binary, exc)
+                self.assertEqual(result, (False, cancelled))
+                entry = report["scenarios"][0]
+                self.assertTrue(entry["status"].startswith(status))
+                self.assertEqual(
+                    entry["error"], str(exc) or "interrupted"
+                )
+
+
 class ExactFactStoreEvaluationTest(unittest.TestCase):
     def test_fixture_preload_requires_the_seeded_fact_in_search_results(self):
         scenario = {

From 1be4cbb841f550a17d7926526a3f88b13353eddf Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:04:07 -0700
Subject: [PATCH 03/12] fix(evals): report profile allocation failures

---
 evals/memory/run_real_model.py      |  6 +++--
 evals/memory/test_run_real_model.py | 39 +++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index 8e549805fb..707a161146 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -230,7 +230,8 @@ def cleanup_eval_artifacts(fixture, eval_env):
         if fixture is not None:
             shutil.rmtree(fixture, ignore_errors=True)
     finally:
-        eval_env.cleanup()
+        if eval_env is not None:
+            eval_env.cleanup()
 
 
 def create_eval_environment(scenario_id):
@@ -819,9 +820,10 @@ def run_scenario(args, scenario, run_dir, report):
     Returns (ok, cancelled). Operational failures and cancellation are recorded
     in the report; the daemon, fixture, and profile are removed on every path.
     """
-    eval_env = create_eval_environment(scenario["id"])
+    eval_env = None
     fixture = None
     try:
+        eval_env = create_eval_environment(scenario["id"])
         eval_env.start_daemon(args.tracedecay_bin)
         fixture = build_fixture(
             scenario, args.tracedecay_bin, eval_env.env, parent=eval_env.root
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index ec8d0f95fb..1ca3beaaed 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -221,6 +221,45 @@ def scenario(self):
             },
         }
 
+    def test_profile_allocation_failure_writes_error_report(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            with (
+                mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                mock.patch.object(
+                    run_real_model, "load_scenarios", return_value=[self.scenario()]
+                ),
+                mock.patch.object(
+                    run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay"
+                ),
+                mock.patch.object(
+                    run_real_model.tempfile, "TemporaryDirectory",
+                    side_effect=OSError("No space left on device"),
+                ),
+                mock.patch.object(run_real_model.EvalEnvironment, "start_daemon") as daemon,
+                mock.patch.object(run_real_model, "drive_hermes") as hermes,
+                mock.patch.object(run_real_model, "drive_cursor_agent") as cursor,
+                contextlib.redirect_stdout(io.StringIO()),
+                contextlib.redirect_stderr(io.StringIO()),
+            ):
+                self.assertEqual(
+                    run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 1
+                )
+            daemon.assert_not_called()
+            hermes.assert_not_called()
+            cursor.assert_not_called()
+            reports = list(Path(tmp).glob("*/report.json"))
+            self.assertEqual(len(reports), 1)
+            report = json.loads(reports[0].read_text())
+            self.assertEqual(report["status"], "fail")
+            self.assertEqual(report["scenarios"], [{
+                "id": "lifecycle",
+                "contract": "stable",
+                "status": "error (setup or agent failure)",
+                "error": "No space left on device",
+                "assertions": [],
+                "transcripts": [],
+            }])
+
     def test_fresh_environment_inits_and_seeds_against_isolated_daemon(self):
         binary = self.make_bin()
         eval_env = run_real_model.create_eval_environment("lifecycle")

From 0c736fceb2af19ba6c5dd249c4d3bde1ccaa8a87 Mon Sep 17 00:00:00 2001
From: Zack Jackson <zackary.l.jackson@gmail.com>
Date: Wed, 21 Oct 2026 00:26:31 -0700
Subject: [PATCH 04/12] fix(evals): own cancelled processes and preserve run
 reports

---
 evals/memory/run_real_model.py      | 203 +++++++++++++++++-----------
 evals/memory/test_run_real_model.py | 190 +++++++++++++++++++++++++-
 scripts/lib/portable_process.py     |  15 ++
 3 files changed, 324 insertions(+), 84 deletions(-)

diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index 707a161146..0506c21a80 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -30,16 +30,20 @@
 import sys
 import tempfile
 import signal
+import traceback
 from dataclasses import dataclass
 from pathlib import Path
 
 EVAL_DIR = Path(__file__).resolve().parent
 REPO_ROOT = EVAL_DIR.parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+from scripts.lib.portable_process import dies_with_this_process, stop_process
+
 SCENARIO_DIR = EVAL_DIR / "scenarios"
 RUNS_DIR = EVAL_DIR / "runs"
 PROCESS_HELPER = REPO_ROOT / "scripts/lib/portable_process.py"
 DAEMON_READY_TIMEOUT_SECONDS = 60
-DAEMON_STOP_GRACE_SECONDS = 5
+PROCESS_STOP_GRACE_SECONDS = 5
 
 DEFAULT_HERMES_DIR = Path.home() / "hermes-agent"
 DEFAULT_MODEL = "gpt-5.4-mini"
@@ -143,13 +147,6 @@ def start_daemon(self, tracedecay_bin, ready_timeout=DAEMON_READY_TIMEOUT_SECOND
         with open(self.daemon_log, "wb") as log:
             self.daemon = subprocess.Popen(
                 [
-                    sys.executable,
-                    "-S",
-                    str(PROCESS_HELPER),
-                    "exec-session",
-                    "--parent-pid",
-                    str(os.getpid()),
-                    "--",
                     tracedecay_bin,
                     "daemon",
                     "run",
@@ -159,8 +156,10 @@ def start_daemon(self, tracedecay_bin, ready_timeout=DAEMON_READY_TIMEOUT_SECOND
                 env=self.env,
                 stdout=log,
                 stderr=subprocess.STDOUT,
+                start_new_session=True,
+                preexec_fn=dies_with_this_process(),
             )
-        ready = subprocess.run(
+        ready = run(
             [
                 sys.executable,
                 "-S",
@@ -173,9 +172,7 @@ def start_daemon(self, tracedecay_bin, ready_timeout=DAEMON_READY_TIMEOUT_SECOND
                 "--timeout",
                 str(ready_timeout),
             ],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
+            check=False,
         )
         if ready.returncode != 0:
             raise EvalSetupError(
@@ -189,36 +186,30 @@ def daemon_log_tail(self):
         return self.daemon_log.read_text(errors="replace")[-4096:] or "(no daemon output captured)"
 
     def stop_daemon(self):
-        daemon, self.daemon = self.daemon, None
-        if daemon is None:
+        if self.daemon is None:
             return
-        subprocess.run(
-            [
-                sys.executable,
-                "-S",
-                str(PROCESS_HELPER),
-                "stop-group",
-                "--pid",
-                str(daemon.pid),
-                "--grace",
-                str(DAEMON_STOP_GRACE_SECONDS),
-            ],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-        )
-        try:
-            daemon.wait(timeout=DAEMON_STOP_GRACE_SECONDS)
-        except subprocess.TimeoutExpired:
-            daemon.kill()
-            daemon.wait()
+        stop_process(self.daemon, PROCESS_STOP_GRACE_SECONDS)
+        self.daemon = None
 
     def cleanup(self):
+        errors = []
         try:
             self.stop_daemon()
-        finally:
-            if self.socket_dir is not None:
-                shutil.rmtree(self.socket_dir, ignore_errors=True)
+        except (Exception, KeyboardInterrupt) as exc:
+            errors.append(exc)
+        if self.socket_dir is not None:
+            try:
+                shutil.rmtree(self.socket_dir)
+            except FileNotFoundError:
+                pass
+            except (Exception, KeyboardInterrupt) as exc:
+                errors.append(exc)
+        try:
             self._temp_dir.cleanup()
+        except (Exception, KeyboardInterrupt) as exc:
+            errors.append(exc)
+        if errors:
+            raise BaseExceptionGroup("isolated environment cleanup failed", errors)
 
 
 class EvalSetupError(RuntimeError):
@@ -226,12 +217,21 @@ class EvalSetupError(RuntimeError):
 
 
 def cleanup_eval_artifacts(fixture, eval_env):
-    try:
-        if fixture is not None:
-            shutil.rmtree(fixture, ignore_errors=True)
-    finally:
-        if eval_env is not None:
+    errors = []
+    if fixture is not None:
+        try:
+            shutil.rmtree(fixture)
+        except FileNotFoundError:
+            pass
+        except (Exception, KeyboardInterrupt) as exc:
+            errors.append(exc)
+    if eval_env is not None:
+        try:
             eval_env.cleanup()
+        except (Exception, KeyboardInterrupt, BaseExceptionGroup) as exc:
+            errors.append(exc)
+    if errors:
+        raise BaseExceptionGroup("eval artifact cleanup failed", errors)
 
 
 def create_eval_environment(scenario_id):
@@ -271,15 +271,16 @@ def resolve_tracedecay_bin(explicit):
     env = dict(os.environ)
     if explicit:
         env["TRACEDECAY_BIN"] = str(explicit)
-    result = subprocess.run(
+    result = run(
         [str(REPO_ROOT / "scripts/build-eval-binary.sh")],
         env=env,
-        stdout=subprocess.PIPE,
-        text=True,
+        stderr=None,
         check=False,
     )
     if result.returncode != 0:
-        sys.exit(f"no tracedecay binary: scripts/build-eval-binary.sh exited {result.returncode}")
+        raise EvalSetupError(
+            f"no tracedecay binary: scripts/build-eval-binary.sh exited {result.returncode}"
+        )
     return result.stdout.strip()
 
 
@@ -304,16 +305,30 @@ def load_scenarios(ids):
     return scenarios
 
 
-def run(cmd, cwd=None, env=None, timeout=None, check=True):
-    result = subprocess.run(
+def run(cmd, cwd=None, env=None, timeout=None, check=True, stderr=subprocess.STDOUT):
+    process = subprocess.Popen(
         cmd,
         cwd=cwd,
         env=env,
         stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
+        stderr=stderr,
         text=True,
-        timeout=timeout,
+        start_new_session=True,
+        preexec_fn=dies_with_this_process(),
     )
+    try:
+        stdout, captured_stderr = process.communicate(timeout=timeout)
+    finally:
+        # A turn, setup command, or build owns its descendants even if its
+        # caller is cancelled or the command leader has already exited.
+        try:
+            stop_process(process, PROCESS_STOP_GRACE_SECONDS)
+        finally:
+            if process.stdout is not None:
+                process.stdout.close()
+            if process.stderr is not None:
+                process.stderr.close()
+    result = subprocess.CompletedProcess(cmd, process.returncode, stdout, captured_stderr)
     if check and result.returncode != 0:
         raise EvalSetupError(
             f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}"
@@ -822,6 +837,8 @@ def run_scenario(args, scenario, run_dir, report):
     """
     eval_env = None
     fixture = None
+    failure = None
+    cancelled = False
     try:
         eval_env = create_eval_environment(scenario["id"])
         eval_env.start_daemon(args.tracedecay_bin)
@@ -834,17 +851,32 @@ def run_scenario(args, scenario, run_dir, report):
             transcripts = drive_cursor_agent(args, scenario, fixture, run_dir, eval_env.env)
         outcomes = evaluate_assertions(scenario, args.tracedecay_bin, fixture, eval_env.env)
     except KeyboardInterrupt as exc:
-        report["scenarios"].append(error_entry(scenario, "cancelled", str(exc) or "interrupted"))
-        print(f"[{scenario['id']}] cancelled")
-        return False, True
+        failure = error_entry(scenario, "cancelled", str(exc) or "interrupted")
+        cancelled = True
     except Exception as exc:
-        report["scenarios"].append(
-            error_entry(scenario, "error (setup or agent failure)", str(exc))
-        )
-        print(f"[{scenario['id']}] error: {exc}", file=sys.stderr)
-        return False, False
+        failure = error_entry(scenario, "error (setup or agent failure)", str(exc))
     finally:
-        cleanup_eval_artifacts(fixture, eval_env)
+        try:
+            cleanup_eval_artifacts(fixture, eval_env)
+        except (Exception, KeyboardInterrupt, BaseExceptionGroup) as exc:
+            detail = "".join(traceback.format_exception(exc))
+            cleanup_cancelled = isinstance(exc, KeyboardInterrupt) or (
+                isinstance(exc, BaseExceptionGroup)
+                and exc.subgroup(KeyboardInterrupt) is not None
+            )
+            cancelled |= cleanup_cancelled
+            if failure is None:
+                status = "cancelled" if cleanup_cancelled else "error (cleanup failure)"
+                failure = error_entry(scenario, status, detail)
+                failure["assertions"] = outcomes
+                failure["transcripts"] = transcripts
+            else:
+                failure["cleanup_error"] = detail
+
+    if failure is not None:
+        report["scenarios"].append(failure)
+        print(f"[{scenario['id']}] {failure['status']}: {failure['error']}", file=sys.stderr)
+        return False, cancelled
 
     failed = [o for o in outcomes if not o["passed"]]
     status = "pass" if not failed else "fail"
@@ -917,36 +949,45 @@ def main(argv):
         print(f"\nblocked report written to {report_path}", file=sys.stderr)
         return 2
 
-    args.tracedecay_bin = resolve_tracedecay_bin(args.tracedecay_bin)
-    report["tracedecay_bin"] = args.tracedecay_bin
+    cancellation_requested = False
 
     def raise_cancelled(signum, _frame):
-        raise KeyboardInterrupt(f"cancelled by signal {signum}")
-
-    previous_term = signal.signal(signal.SIGTERM, raise_cancelled)
+        nonlocal cancellation_requested
+        if not cancellation_requested:
+            cancellation_requested = True
+            raise KeyboardInterrupt(f"cancelled by signal {signum}")
+
+    previous_handlers = {
+        sig: signal.signal(sig, raise_cancelled)
+        for sig in (signal.SIGINT, signal.SIGTERM)
+    }
     overall_ok = True
     cancelled = False
     try:
-        for scenario in scenarios:
-            ok, cancelled = run_scenario(args, scenario, run_dir, report)
-            overall_ok &= ok
-            if cancelled:
-                break
-    finally:
-        signal.signal(signal.SIGTERM, previous_term)
-
-    if cancelled:
-        report["status"] = "cancelled"
+        try:
+            args.tracedecay_bin = resolve_tracedecay_bin(args.tracedecay_bin)
+            report["tracedecay_bin"] = args.tracedecay_bin
+            for scenario in scenarios:
+                ok, cancelled = run_scenario(args, scenario, run_dir, report)
+                overall_ok &= ok
+                if cancelled:
+                    break
+        except KeyboardInterrupt as exc:
+            cancelled = True
+            report["error"] = str(exc) or "interrupted"
+        except Exception as exc:
+            overall_ok = False
+            report["error"] = str(exc)
+
+        cancelled |= cancellation_requested
+        report["status"] = "cancelled" if cancelled else "pass" if overall_ok else "fail"
         report_path = run_dir / "report.json"
         report_path.write_text(json.dumps(report, indent=2) + "\n")
-        print(f"\ncancelled; report written to {report_path}", file=sys.stderr)
-        return 130
-
-    report["status"] = "pass" if overall_ok else "fail"
-    report_path = run_dir / "report.json"
-    report_path.write_text(json.dumps(report, indent=2) + "\n")
-    print(f"\nreport written to {report_path}")
-    return 0 if overall_ok else 1
+        print(f"\n{report['status']}; report written to {report_path}")
+        return 130 if cancelled else 0 if overall_ok else 1
+    finally:
+        for sig, handler in previous_handlers.items():
+            signal.signal(sig, handler)
 
 
 if __name__ == "__main__":
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index 1ca3beaaed..f0baad7ca2 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -4,8 +4,11 @@
 import io
 import json
 import os
+import signal
 import subprocess
+import sys
 import tempfile
+import time
 import unittest
 from unittest import mock
 from pathlib import Path
@@ -15,6 +18,7 @@
 SPEC = importlib.util.spec_from_file_location("run_real_model", MODULE_PATH)
 run_real_model = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(run_real_model)
+from scripts.lib import portable_process
 
 BENCHMARK_PATH = MODULE_PATH.parent.parent.parent / "benchmark_data" / "run_benchmarks.py"
 BENCHMARK_SPEC = importlib.util.spec_from_file_location("run_benchmarks", BENCHMARK_PATH)
@@ -108,7 +112,7 @@ def test_explicit_binary_is_used_and_must_exist(self):
                 self.assertEqual(
                     run_real_model.resolve_tracedecay_bin(binary), str(binary)
                 )
-                with self.assertRaises(SystemExit):
+                with self.assertRaises(run_real_model.EvalSetupError):
                     run_real_model.resolve_tracedecay_bin(Path(tmp) / "missing")
 
     def test_default_never_falls_back_to_a_prebuilt_or_path_binary(self):
@@ -119,11 +123,27 @@ def test_default_never_falls_back_to_a_prebuilt_or_path_binary(self):
             path_binary.chmod(0o755)
             with (
                 mock.patch.dict("os.environ", {"PATH": tmp}),
-                mock.patch.object(run_real_model.subprocess, "run", return_value=failed),
-                self.assertRaises(SystemExit),
+                mock.patch.object(run_real_model, "run", return_value=failed),
+                self.assertRaises(run_real_model.EvalSetupError),
             ):
                 run_real_model.resolve_tracedecay_bin(None)
 
+    def test_build_failure_writes_a_failed_report_before_any_scenario_starts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            with (
+                mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                mock.patch.object(run_real_model, "load_scenarios", return_value=[{"id": "test"}]),
+                mock.patch.object(run_real_model, "run", return_value=subprocess.CompletedProcess([], 2, stdout="")),
+                mock.patch.object(run_real_model, "run_scenario") as scenario,
+                contextlib.redirect_stdout(io.StringIO()),
+            ):
+                self.assertEqual(run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 1)
+            report = json.loads(next(Path(tmp).glob("*/report.json")).read_text())
+        scenario.assert_not_called()
+        self.assertEqual(report["status"], "fail")
+        self.assertIn("build-eval-binary.sh exited 2", report["error"])
+        self.assertIsNone(report["tracedecay_bin"])
+
 
 class CursorComposerSelectionTest(unittest.TestCase):
     def test_cursor_driver_loads_the_installed_plugin_in_composer(self):
@@ -328,6 +348,170 @@ def test_agent_failure_and_cancellation_write_error_entries(self):
                     entry["error"], str(exc) or "interrupted"
                 )
 
+    def test_daemon_cleanup_failure_preserves_report_and_removes_artifacts(self):
+        binary = self.make_bin()
+        for agent_failure, cleanup_failure in (
+            (RuntimeError("agent failed"), OSError("cleanup failed")),
+            (None, OSError("cleanup failed")),
+            (None, KeyboardInterrupt("cancelled during cleanup")),
+        ):
+            with self.subTest(agent=agent_failure, cleanup=cleanup_failure):
+                environment = run_real_model.create_eval_environment("cleanup-failure")
+                daemon_pid = None
+                outcomes = [{"name": "completed assertion", "passed": True}]
+                transcripts = [{"turn_valid": True}]
+
+                def stop_then_fail(env):
+                    nonlocal daemon_pid
+                    daemon_pid = env.daemon.pid
+                    run_real_model.stop_process(env.daemon, 5)
+                    raise cleanup_failure
+
+                with tempfile.TemporaryDirectory() as tmp:
+                    with (
+                        mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                        mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]),
+                        mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=binary),
+                        mock.patch.object(run_real_model, "create_eval_environment", return_value=environment),
+                        mock.patch.object(run_real_model, "drive_hermes", side_effect=agent_failure, return_value=transcripts),
+                        mock.patch.object(run_real_model, "evaluate_assertions", return_value=outcomes),
+                        mock.patch.object(run_real_model.EvalEnvironment, "stop_daemon", autospec=True, side_effect=stop_then_fail),
+                        contextlib.redirect_stdout(io.StringIO()),
+                        contextlib.redirect_stderr(io.StringIO()),
+                    ):
+                        cancelled = isinstance(cleanup_failure, KeyboardInterrupt)
+                        self.assertEqual(
+                            run_real_model.main(["--agent-turn", "--i-understand-model-cost"]),
+                            130 if cancelled else 1,
+                        )
+                    report = json.loads(next(Path(tmp).glob("*/report.json")).read_text())
+                self.assertEqual(report["status"], "cancelled" if cancelled else "fail")
+                entry = report["scenarios"][0]
+                if agent_failure is not None:
+                    self.assertEqual(entry["error"], str(agent_failure))
+                    self.assertIn(str(cleanup_failure), entry["cleanup_error"])
+                else:
+                    self.assertIn(str(cleanup_failure), entry["error"])
+                    self.assertEqual(entry["assertions"], outcomes)
+                    self.assertEqual(entry["transcripts"], transcripts)
+                self.assertFalse(environment.root.exists())
+                self.assertFalse(environment.socket_dir.exists())
+                with self.assertRaises(ProcessLookupError):
+                    os.kill(daemon_pid, 0)
+
+
+class ProcessCancellationTest(unittest.TestCase):
+    def test_stop_failure_still_kills_and_reaps_the_owned_process(self):
+        process = subprocess.Popen(
+            [sys.executable, "-c", "import time; time.sleep(600)"],
+            start_new_session=True,
+        )
+        try:
+            with mock.patch.object(portable_process, "_stop_group", side_effect=OSError("stop failed")):
+                with self.assertRaisesRegex(OSError, "stop failed"):
+                    portable_process.stop_process(process, 1)
+            self.assertIsNotNone(process.returncode)
+            with self.assertRaises(ProcessLookupError):
+                os.kill(process.pid, 0)
+        finally:
+            portable_process.stop_process(process, 1)
+
+    def test_signals_during_binary_resolution_and_turns_stop_owned_groups_and_write_reports(self):
+        workload = """
+import os, subprocess, sys, time
+from pathlib import Path
+child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(600)'])
+Path(sys.argv[1]).write_text(str(os.getpid()))
+time.sleep(600)
+"""
+        entrypoint = """
+import sys
+from pathlib import Path
+from unittest import mock
+sys.path.insert(0, sys.argv[1])
+import run_real_model as runner
+root, phase = Path(sys.argv[2]), sys.argv[3]
+runner.RUNS_DIR = root / 'runs'
+scenario = {'id': 'cancel', 'setup': {}, 'assertions': []}
+def work(*args, **kwargs):
+    return runner.run([sys.executable, '-c', sys.argv[4], str(root / 'work.pid')])
+with mock.patch.object(runner, 'load_scenarios', return_value=[scenario]):
+    if phase == 'build':
+        runner.resolve_tracedecay_bin = work
+    else:
+        runner.resolve_tracedecay_bin = lambda explicit: sys.argv[5]
+        runner.drive_hermes = work
+    sys.exit(runner.main(['--agent-turn', '--i-understand-model-cost']))
+"""
+        for phase in ("build", "turn"):
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                with self.subTest(phase=phase, signal=sig), tempfile.TemporaryDirectory() as tmp:
+                    root = Path(tmp)
+                    binary = root / "tracedecay"
+                    binary.write_text(FAKE_BIN)
+                    binary.chmod(0o755)
+                    process = subprocess.Popen(
+                        [sys.executable, "-c", entrypoint, str(MODULE_PATH.parent), tmp,
+                         phase, workload, str(binary)],
+                        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
+                        start_new_session=True,
+                    )
+                    group = None
+                    try:
+                        deadline = time.monotonic() + 15
+                        while not (root / "work.pid").exists() and process.poll() is None:
+                            if time.monotonic() >= deadline:
+                                self.fail("owned workload did not start")
+                            time.sleep(0.02)
+                        self.assertTrue((root / "work.pid").exists())
+                        group = int((root / "work.pid").read_text())
+                        process.send_signal(sig)
+                        output, _ = process.communicate(timeout=15)
+                        self.assertEqual(process.returncode, 130, output)
+                        report = json.loads(next((root / "runs").glob("*/report.json")).read_text())
+                        self.assertEqual(report["status"], "cancelled", report)
+                        if phase == "turn":
+                            self.assertEqual(report["scenarios"][0]["status"], "cancelled")
+                        self.assertEqual(
+                            subprocess.run(
+                                [sys.executable, "-S", str(run_real_model.PROCESS_HELPER),
+                                 "group-alive", "--pid", str(group)],
+                                check=False,
+                            ).returncode,
+                            1,
+                            "a cancelled workload left runnable descendants",
+                        )
+                        with self.assertRaises(ProcessLookupError):
+                            os.kill(group, 0)
+                    finally:
+                        run_real_model.stop_process(process, 1)
+                        if group is not None:
+                            try:
+                                os.killpg(group, signal.SIGKILL)
+                            except ProcessLookupError:
+                                pass
+                        if process.stdout is not None:
+                            process.stdout.close()
+
+    def test_timeout_stops_descendants_and_reaps_the_leader(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pid_path = Path(tmp) / "pid"
+            with self.assertRaises(subprocess.TimeoutExpired):
+                run_real_model.run(
+                    [sys.executable, "-c",
+                     "import os, subprocess, sys, time; "
+                     "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(600)']); "
+                     "open(sys.argv[1], 'w').write(str(os.getpid())); time.sleep(600)",
+                     str(pid_path)], timeout=0.5,
+                )
+            pid = int(pid_path.read_text())
+            with self.assertRaises(ProcessLookupError):
+                os.kill(pid, 0)
+            self.assertEqual(subprocess.run(
+                [sys.executable, "-S", str(run_real_model.PROCESS_HELPER),
+                 "group-alive", "--pid", str(pid)], check=False,
+            ).returncode, 1)
+
 
 class ExactFactStoreEvaluationTest(unittest.TestCase):
     def test_fixture_preload_requires_the_seeded_fact_in_search_results(self):
diff --git a/scripts/lib/portable_process.py b/scripts/lib/portable_process.py
index 718a9f0f8e..29735bab8c 100755
--- a/scripts/lib/portable_process.py
+++ b/scripts/lib/portable_process.py
@@ -165,6 +165,21 @@ def _stop_group(pid: int, grace_seconds: float) -> bool:
     return forced
 
 
+def stop_process(process: subprocess.Popen, grace_seconds: float) -> None:
+    """Stop an owned session and reap its leader, including on stop errors."""
+    try:
+        _stop_group(process.pid, grace_seconds)
+    except BaseException:
+        _signal_group(process.pid, signal.SIGKILL)
+        raise
+    finally:
+        try:
+            process.wait(timeout=grace_seconds)
+        except subprocess.TimeoutExpired:
+            _signal_group(process.pid, signal.SIGKILL)
+            process.wait()
+
+
 def _return_code(status: int) -> int:
     return status if status >= 0 else 128 + abs(status)
 

From f934c5e855c486eda745a928bef0b842ade9d23a Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 00:42:19 -0700
Subject: [PATCH 05/12] fix(evals): retain process ownership through
 cancellation and cleanup

Use the shared process-group owner for commands and daemon shutdown, preserving the original error and a cancelled report through setup, execution, cleanup, and report publication.

Allocate fixture ownership before startup and seeding. Retain profile and fixture inputs until every owned process is confirmed stopped, and exercise cancellation and cleanup failures with real child processes.
---
 evals/memory/run_real_model.py      | 334 ++++++++------
 evals/memory/test_run_real_model.py | 664 ++++++++++++++++++++++++++--
 scripts/lib/portable_process.py     |  45 +-
 3 files changed, 862 insertions(+), 181 deletions(-)

diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index 0506c21a80..8ee38b6eca 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -25,19 +25,19 @@
 import os
 import re
 import shutil
+import signal
 import sqlite3
 import subprocess
 import sys
 import tempfile
-import signal
-import traceback
 from dataclasses import dataclass
 from pathlib import Path
 
 EVAL_DIR = Path(__file__).resolve().parent
 REPO_ROOT = EVAL_DIR.parent.parent
-sys.path.insert(0, str(REPO_ROOT))
-from scripts.lib.portable_process import dies_with_this_process, stop_process
+sys.path.insert(0, str(REPO_ROOT / "scripts/lib"))
+
+from portable_process import dies_with_this_process, stop_owned_process
 
 SCENARIO_DIR = EVAL_DIR / "scenarios"
 RUNS_DIR = EVAL_DIR / "runs"
@@ -130,7 +130,6 @@ class EvalEnvironment:
     data_dir: Path
     global_db: Path
     env: dict[str, str]
-    _temp_dir: tempfile.TemporaryDirectory
     socket_dir: Path | None = None
     daemon: subprocess.Popen | None = None
     daemon_log: Path | None = None
@@ -188,61 +187,71 @@ def daemon_log_tail(self):
     def stop_daemon(self):
         if self.daemon is None:
             return
-        stop_process(self.daemon, PROCESS_STOP_GRACE_SECONDS)
+        # Teardown must not depend on launching another Python executable.
+        # Retain ownership until the canonical group stop and reap complete.
+        stop_owned_process(self.daemon, PROCESS_STOP_GRACE_SECONDS)
         self.daemon = None
 
     def cleanup(self):
+        self.stop_daemon()
         errors = []
-        try:
-            self.stop_daemon()
-        except (Exception, KeyboardInterrupt) as exc:
-            errors.append(exc)
-        if self.socket_dir is not None:
+        for path in (self.socket_dir, self.root):
+            if path is None:
+                continue
             try:
-                shutil.rmtree(self.socket_dir)
+                shutil.rmtree(path)
             except FileNotFoundError:
                 pass
-            except (Exception, KeyboardInterrupt) as exc:
+            except Exception as exc:
                 errors.append(exc)
-        try:
-            self._temp_dir.cleanup()
-        except (Exception, KeyboardInterrupt) as exc:
-            errors.append(exc)
+        if len(errors) == 1:
+            raise errors[0]
         if errors:
-            raise BaseExceptionGroup("isolated environment cleanup failed", errors)
+            raise ExceptionGroup(
+                "isolated environment cleanup failed: "
+                + "; ".join(str(error) for error in errors),
+                errors,
+            )
 
 
 class EvalSetupError(RuntimeError):
     """An operational failure (daemon, init, install) reported instead of exiting."""
 
 
-def cleanup_eval_artifacts(fixture, eval_env):
-    errors = []
-    if fixture is not None:
-        try:
-            shutil.rmtree(fixture)
-        except FileNotFoundError:
-            pass
-        except (Exception, KeyboardInterrupt) as exc:
-            errors.append(exc)
-    if eval_env is not None:
-        try:
-            eval_env.cleanup()
-        except (Exception, KeyboardInterrupt, BaseExceptionGroup) as exc:
-            errors.append(exc)
-    if errors:
-        raise BaseExceptionGroup("eval artifact cleanup failed", errors)
+class OwnedProcessCleanupError(EvalSetupError):
+    """Transfer an unconfirmed command stop to the enclosing resource owner."""
+
+    def __init__(self, process, original_error, cleanup_error):
+        self.process = process
+        self.original_error = original_error
+        self.cleanup_error = cleanup_error
+        super().__init__(f"owned process group {process.pid} cleanup failed: {cleanup_error}")
+
+    @property
+    def cancelled(self):
+        return isinstance(self.original_error, KeyboardInterrupt) or isinstance(
+            self.cleanup_error, KeyboardInterrupt
+        )
+
+    def primary_detail(self):
+        error = self.original_error if self.original_error is not None else self.cleanup_error
+        return str(error) or ("interrupted" if self.cancelled else type(error).__name__)
 
 
 def create_eval_environment(scenario_id):
     safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario_id)
-    temp_dir = tempfile.TemporaryDirectory(prefix=f"tracedecay-eval-env-{safe_id}-")
-    root = Path(temp_dir.name)
+    # Explicit ownership prevents an implicit temporary-directory finalizer
+    # from deleting the profile if stopping an owned daemon fails.
+    root = Path(tempfile.mkdtemp(prefix=f"tracedecay-eval-env-{safe_id}-"))
     home = root / "home"
     data_dir = root / ".tracedecay"
     global_db = data_dir / "global.db"
-    home.mkdir(parents=True, exist_ok=True)
-    data_dir.mkdir(parents=True, exist_ok=True)
+    try:
+        home.mkdir(parents=True, exist_ok=True)
+        data_dir.mkdir(parents=True, exist_ok=True)
+    except BaseException:
+        shutil.rmtree(root, ignore_errors=True)
+        raise
 
     env = dict(os.environ)
     env["HOME"] = str(home)
@@ -258,7 +267,6 @@ def create_eval_environment(scenario_id):
         data_dir=data_dir,
         global_db=global_db,
         env=env,
-        _temp_dir=temp_dir,
     )
 
 
@@ -274,8 +282,8 @@ def resolve_tracedecay_bin(explicit):
     result = run(
         [str(REPO_ROOT / "scripts/build-eval-binary.sh")],
         env=env,
-        stderr=None,
         check=False,
+        stderr=None,
     )
     if result.returncode != 0:
         raise EvalSetupError(
@@ -316,23 +324,34 @@ def run(cmd, cwd=None, env=None, timeout=None, check=True, stderr=subprocess.STD
         start_new_session=True,
         preexec_fn=dies_with_this_process(),
     )
+    original_error = None
     try:
-        stdout, captured_stderr = process.communicate(timeout=timeout)
-    finally:
-        # A turn, setup command, or build owns its descendants even if its
-        # caller is cancelled or the command leader has already exited.
-        try:
-            stop_process(process, PROCESS_STOP_GRACE_SECONDS)
-        finally:
-            if process.stdout is not None:
-                process.stdout.close()
-            if process.stderr is not None:
-                process.stderr.close()
-    result = subprocess.CompletedProcess(cmd, process.returncode, stdout, captured_stderr)
-    if check and result.returncode != 0:
-        raise EvalSetupError(
-            f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}"
+        output, captured_stderr = process.communicate(timeout=timeout)
+        result = subprocess.CompletedProcess(
+            cmd, process.returncode, stdout=output, stderr=captured_stderr
         )
+        if check and result.returncode != 0:
+            raise EvalSetupError(
+                f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}"
+            )
+    except BaseException as exc:
+        original_error = exc
+    try:
+        # Popen's context manager waits during an exception without stopping
+        # the child. Explicit ownership makes cancellation and timeouts stop
+        # the entire turn, including its MCP children, before fixture cleanup.
+        stop_owned_process(process, PROCESS_STOP_GRACE_SECONDS)
+    except BaseException as exc:
+        # A signal can arrive for the first time during the stop itself. Keep
+        # the Popen reachable so the caller retries before deleting its files.
+        raise OwnedProcessCleanupError(process, original_error, exc) from original_error
+    finally:
+        if process.stdout is not None:
+            process.stdout.close()
+        if process.stderr is not None:
+            process.stderr.close()
+    if original_error is not None:
+        raise original_error
     return result
 
 
@@ -371,23 +390,6 @@ def require_fact_record(response, operation):
     return fact
 
 
-def build_fixture(scenario, tracedecay_bin, env, parent=None):
-    """Creates and seeds the fixture; removes it again if any setup step fails."""
-    safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario["id"])
-    fixture = Path(
-        tempfile.mkdtemp(
-            prefix=f"tracedecay-eval-{safe_id}-",
-            dir=str(parent) if parent is not None else None,
-        )
-    )
-    try:
-        seed_fixture(scenario, tracedecay_bin, env, fixture)
-    except BaseException:
-        shutil.rmtree(fixture, ignore_errors=True)
-        raise
-    return fixture
-
-
 def seed_fixture(scenario, tracedecay_bin, env, fixture):
     (fixture / "src").mkdir()
     (fixture / "src/lib.rs").write_text("pub fn eval_fixture_marker() {}\n")
@@ -833,69 +835,106 @@ def run_scenario(args, scenario, run_dir, report):
     """Runs one scenario in an owned profile/daemon/fixture scope.
 
     Returns (ok, cancelled). Operational failures and cancellation are recorded
-    in the report; the daemon, fixture, and profile are removed on every path.
+    in the report. Cleanup stops owned processes before removing their files;
+    a cleanup failure preserves both its error and the original result.
     """
     eval_env = None
     fixture = None
-    failure = None
+    entry = error_entry(scenario, "error (setup or agent failure)", "scenario did not finish")
+    report["scenarios"].append(entry)
+    ok = False
     cancelled = False
+    pending_command = None
+
+    def cleanup():
+        nonlocal pending_command
+        if pending_command is not None:
+            stop_owned_process(pending_command, PROCESS_STOP_GRACE_SECONDS)
+            pending_command = None
+        if eval_env is not None:
+            eval_env.cleanup()
+
     try:
         eval_env = create_eval_environment(scenario["id"])
-        eval_env.start_daemon(args.tracedecay_bin)
-        fixture = build_fixture(
-            scenario, args.tracedecay_bin, eval_env.env, parent=eval_env.root
+        safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario["id"])
+        # This scope owns the fixture before any daemon or seeding command can
+        # use it. Setup errors therefore follow the same stop-before-delete
+        # order as completed turns and cancellation.
+        fixture = Path(
+            tempfile.mkdtemp(prefix=f"tracedecay-eval-{safe_id}-", dir=eval_env.root)
         )
+        eval_env.start_daemon(args.tracedecay_bin)
+        seed_fixture(scenario, args.tracedecay_bin, eval_env.env, fixture)
         if args.driver == "hermes":
             transcripts = drive_hermes(args, scenario, fixture, run_dir, eval_env.env)
         else:
             transcripts = drive_cursor_agent(args, scenario, fixture, run_dir, eval_env.env)
         outcomes = evaluate_assertions(scenario, args.tracedecay_bin, fixture, eval_env.env)
+        failed = [o for o in outcomes if not o["passed"]]
+        status = "pass" if not failed else "fail"
+        if failed and scenario.get("contract") == "pending-sibling":
+            status = "fail (note: scenario contract is pending-sibling, see contract_notes)"
+        if not all(t["turn_valid"] for t in transcripts):
+            status = "error (agent turn invalid, see transcript logs)"
+            failed = failed or [{"name": "agent-turn", "passed": False}]
+        entry.pop("error")
+        entry.update(status=status, assertions=outcomes, transcripts=transcripts)
+        ok = not failed
+    except OwnedProcessCleanupError as exc:
+        pending_command = exc.process
+        cancelled = exc.cancelled
+        if cancelled:
+            entry["status"] = "cancelled"
+        entry["error"] = exc.primary_detail()
     except KeyboardInterrupt as exc:
-        failure = error_entry(scenario, "cancelled", str(exc) or "interrupted")
+        entry.update(status="cancelled", error=str(exc) or "interrupted")
         cancelled = True
     except Exception as exc:
-        failure = error_entry(scenario, "error (setup or agent failure)", str(exc))
+        entry["error"] = str(exc)
     finally:
         try:
-            cleanup_eval_artifacts(fixture, eval_env)
-        except (Exception, KeyboardInterrupt, BaseExceptionGroup) as exc:
-            detail = "".join(traceback.format_exception(exc))
-            cleanup_cancelled = isinstance(exc, KeyboardInterrupt) or (
-                isinstance(exc, BaseExceptionGroup)
-                and exc.subgroup(KeyboardInterrupt) is not None
-            )
-            cancelled |= cleanup_cancelled
-            if failure is None:
-                status = "cancelled" if cleanup_cancelled else "error (cleanup failure)"
-                failure = error_entry(scenario, status, detail)
-                failure["assertions"] = outcomes
-                failure["transcripts"] = transcripts
-            else:
-                failure["cleanup_error"] = detail
-
-    if failure is not None:
-        report["scenarios"].append(failure)
-        print(f"[{scenario['id']}] {failure['status']}: {failure['error']}", file=sys.stderr)
-        return False, cancelled
-
-    failed = [o for o in outcomes if not o["passed"]]
-    status = "pass" if not failed else "fail"
-    if failed and scenario.get("contract") == "pending-sibling":
-        status = "fail (note: scenario contract is pending-sibling, see contract_notes)"
-    if not all(t["turn_valid"] for t in transcripts):
-        status = "error (agent turn invalid, see transcript logs)"
-        failed = failed or [{"name": "agent-turn", "passed": False}]
-    report["scenarios"].append(
-        {
-            "id": scenario["id"],
-            "contract": scenario.get("contract", "stable"),
-            "status": status,
-            "assertions": outcomes,
-            "transcripts": transcripts,
-        }
-    )
-    print(f"[{scenario['id']}] {status}")
-    for outcome in outcomes:
+            try:
+                cleanup()
+            except KeyboardInterrupt as exc:
+                # The first signal may arrive during teardown. main keeps
+                # subsequent signals from interrupting this final cleanup.
+                cancelled = True
+                ok = False
+                entry["status"] = "cancelled"
+                entry.setdefault("error", str(exc) or "interrupted")
+                cleanup()
+        except BaseException as exc:
+            ok = False
+            cancelled |= isinstance(exc, KeyboardInterrupt)
+            entry["cleanup_error"] = str(exc) or type(exc).__name__
+            if pending_command is not None:
+                entry["command_pid"] = pending_command.pid
+            for label, path in (
+                ("retained_fixture", fixture),
+                ("retained_profile", eval_env.root if eval_env is not None else None),
+                ("retained_socket_dir", eval_env.socket_dir if eval_env is not None else None),
+            ):
+                if path is None:
+                    continue
+                try:
+                    path.stat()
+                except FileNotFoundError:
+                    continue
+                except OSError:
+                    # An inaccessible owned path may still exist. Preserve
+                    # its location without replacing the cleanup failure.
+                    pass
+                entry[label] = str(path)
+            if eval_env is not None and eval_env.daemon is not None:
+                entry["daemon_pid"] = eval_env.daemon.pid
+            entry["status"] = "cancelled" if cancelled else "error (cleanup failure)"
+
+    print(f"[{scenario['id']}] {entry['status']}")
+    if entry.get("error"):
+        print(f"  {entry['error']}", file=sys.stderr)
+    if entry.get("cleanup_error"):
+        print(f"  cleanup: {entry['cleanup_error']}", file=sys.stderr)
+    for outcome in entry["assertions"]:
         marker = "pass" if outcome["passed"] else "FAIL"
         if outcome.get("error"):
             print(f"  [{marker}] {outcome['name']}, {outcome['error']}")
@@ -904,7 +943,7 @@ def run_scenario(args, scenario, run_dir, report):
                 f"  [{marker}] {outcome['name']}, actual {outcome['actual']} "
                 f"{outcome['op']} expected {outcome['expected']}"
             )
-    return not failed, False
+    return ok, cancelled
 
 
 def error_entry(scenario, status, error):
@@ -949,12 +988,11 @@ def main(argv):
         print(f"\nblocked report written to {report_path}", file=sys.stderr)
         return 2
 
-    cancellation_requested = False
-
+    cancel_requested = False
     def raise_cancelled(signum, _frame):
-        nonlocal cancellation_requested
-        if not cancellation_requested:
-            cancellation_requested = True
+        nonlocal cancel_requested
+        if not cancel_requested:
+            cancel_requested = True
             raise KeyboardInterrupt(f"cancelled by signal {signum}")
 
     previous_handlers = {
@@ -963,32 +1001,62 @@ def raise_cancelled(signum, _frame):
     }
     overall_ok = True
     cancelled = False
+    phase = "binary resolution"
+    report_path = run_dir / "report.json"
     try:
         try:
             args.tracedecay_bin = resolve_tracedecay_bin(args.tracedecay_bin)
             report["tracedecay_bin"] = args.tracedecay_bin
+            phase = "scenario execution"
             for scenario in scenarios:
                 ok, cancelled = run_scenario(args, scenario, run_dir, report)
                 overall_ok &= ok
+                cancelled |= cancel_requested
                 if cancelled:
                     break
-        except KeyboardInterrupt as exc:
-            cancelled = True
-            report["error"] = str(exc) or "interrupted"
+        except OwnedProcessCleanupError as exc:
+            overall_ok = False
+            cancelled = exc.cancelled
+            report["reason"] = exc.primary_detail()
+            report["phase"] = phase
+            # Binary resolution runs before any environment exists, so main
+            # is the owner of a command whose initial stop did not complete.
+            try:
+                try:
+                    stop_owned_process(exc.process, PROCESS_STOP_GRACE_SECONDS)
+                except KeyboardInterrupt:
+                    cancelled = True
+                    stop_owned_process(exc.process, PROCESS_STOP_GRACE_SECONDS)
+            except BaseException as cleanup_error:
+                cancelled |= isinstance(cleanup_error, KeyboardInterrupt)
+                report["cleanup_error"] = str(cleanup_error) or type(cleanup_error).__name__
+                report["command_pid"] = exc.process.pid
         except Exception as exc:
             overall_ok = False
-            report["error"] = str(exc)
+            report["reason"] = str(exc)
+            report["phase"] = phase
 
-        cancelled |= cancellation_requested
-        report["status"] = "cancelled" if cancelled else "pass" if overall_ok else "fail"
-        report_path = run_dir / "report.json"
+        cancelled |= cancel_requested
+        report["status"] = "cancelled" if cancelled else ("pass" if overall_ok else "fail")
+        phase = "report publication"
+        report_path.write_text(json.dumps(report, indent=2) + "\n")
+    except KeyboardInterrupt as exc:
+        # This scope covers resolution, gaps between scenarios, and report
+        # publication. Finish the report before restoring signal handlers.
+        cancelled = True
+        report.update(status="cancelled", reason=str(exc) or "interrupted", phase=phase)
         report_path.write_text(json.dumps(report, indent=2) + "\n")
-        print(f"\n{report['status']}; report written to {report_path}")
-        return 130 if cancelled else 0 if overall_ok else 1
     finally:
         for sig, handler in previous_handlers.items():
             signal.signal(sig, handler)
 
+    if cancelled:
+        print(f"\ncancelled; report written to {report_path}", file=sys.stderr)
+        return 130
+
+    print(f"\nreport written to {report_path}")
+    return 0 if overall_ok else 1
+
 
 if __name__ == "__main__":
     sys.exit(main(sys.argv[1:]))
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index f0baad7ca2..c98dcb70e0 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -1,5 +1,6 @@
 import argparse
 import contextlib
+import errno
 import importlib.util
 import io
 import json
@@ -8,6 +9,7 @@
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 from unittest import mock
@@ -18,7 +20,8 @@
 SPEC = importlib.util.spec_from_file_location("run_real_model", MODULE_PATH)
 run_real_model = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(run_real_model)
-from scripts.lib import portable_process
+
+import portable_process
 
 BENCHMARK_PATH = MODULE_PATH.parent.parent.parent / "benchmark_data" / "run_benchmarks.py"
 BENCHMARK_SPEC = importlib.util.spec_from_file_location("run_benchmarks", BENCHMARK_PATH)
@@ -128,6 +131,7 @@ def test_default_never_falls_back_to_a_prebuilt_or_path_binary(self):
             ):
                 run_real_model.resolve_tracedecay_bin(None)
 
+
     def test_build_failure_writes_a_failed_report_before_any_scenario_starts(self):
         with tempfile.TemporaryDirectory() as tmp:
             with (
@@ -141,7 +145,7 @@ def test_build_failure_writes_a_failed_report_before_any_scenario_starts(self):
             report = json.loads(next(Path(tmp).glob("*/report.json")).read_text())
         scenario.assert_not_called()
         self.assertEqual(report["status"], "fail")
-        self.assertIn("build-eval-binary.sh exited 2", report["error"])
+        self.assertIn("build-eval-binary.sh exited 2", report["reason"])
         self.assertIsNone(report["tracedecay_bin"])
 
 
@@ -188,15 +192,15 @@ def test_eval_environment_pins_home_and_profile_storage_to_tempdir(self):
         self.assertNotEqual(env.global_db, Path.home() / ".tracedecay/global.db")
 
     def test_cleanup_removes_fixture_and_isolated_eval_store(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            fixture = Path(tmp) / "fixture"
-            fixture.mkdir()
-            eval_env = run_real_model.create_eval_environment("cleanup-store")
+        eval_env = run_real_model.create_eval_environment("cleanup-store")
+        self.addCleanup(eval_env.cleanup)
+        fixture = eval_env.root / "fixture"
+        fixture.mkdir()
 
-            run_real_model.cleanup_eval_artifacts(fixture, eval_env)
+        eval_env.cleanup()
 
-            self.assertFalse(fixture.exists())
-            self.assertFalse(eval_env.root.exists())
+        self.assertFalse(fixture.exists())
+        self.assertFalse(eval_env.root.exists())
 
 
 FAKE_BIN = """#!/usr/bin/env python3
@@ -252,7 +256,7 @@ def test_profile_allocation_failure_writes_error_report(self):
                     run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay"
                 ),
                 mock.patch.object(
-                    run_real_model.tempfile, "TemporaryDirectory",
+                    run_real_model.tempfile, "mkdtemp",
                     side_effect=OSError("No space left on device"),
                 ),
                 mock.patch.object(run_real_model.EvalEnvironment, "start_daemon") as daemon,
@@ -284,32 +288,56 @@ def test_fresh_environment_inits_and_seeds_against_isolated_daemon(self):
         binary = self.make_bin()
         eval_env = run_real_model.create_eval_environment("lifecycle")
         try:
+            fixture = eval_env.root / "fixture"
+            fixture.mkdir()
             eval_env.start_daemon(binary, ready_timeout=15)
-            fixture = run_real_model.build_fixture(
-                self.scenario(), binary, eval_env.env, parent=eval_env.root
-            )
+            run_real_model.seed_fixture(self.scenario(), binary, eval_env.env, fixture)
             self.assertTrue(fixture.is_relative_to(eval_env.root))
             pid = int((eval_env.data_dir / "daemon.pid").read_text())
         finally:
-            run_real_model.cleanup_eval_artifacts(None, eval_env)
+            eval_env.cleanup()
         self.assertFalse(eval_env.root.exists())
         self.assertFalse(Path(eval_env.env["TRACEDECAY_DAEMON_SOCKET"]).parent.exists())
         with self.assertRaises(ProcessLookupError):
             os.kill(pid, 0)
 
-    def test_init_failure_removes_fixture_and_raises_typed_error(self):
+    def test_init_failure_stops_daemon_before_removing_fixture(self):
         binary = self.make_bin()
         eval_env = run_real_model.create_eval_environment("lifecycle")
         self.addCleanup(eval_env.cleanup)
-        eval_env.start_daemon(binary, ready_timeout=15)
         eval_env.env["FAKE_INIT_FAIL"] = "1"
-        with self.assertRaises(run_real_model.EvalSetupError):
-            run_real_model.build_fixture(
-                self.scenario(), binary, eval_env.env, parent=eval_env.root
+        args = argparse.Namespace(tracedecay_bin=binary, driver="hermes")
+        report = {"scenarios": []}
+        removed = []
+        original_remove = run_real_model.shutil.rmtree
+
+        def remove(path, *args, **kwargs):
+            path = Path(path)
+            if path.is_relative_to(eval_env.root) and path.exists():
+                self.assertIsNone(eval_env.daemon, "owned work must stop before deleting inputs")
+                pid = int((eval_env.data_dir / "daemon.pid").read_text())
+                with self.assertRaises(ProcessLookupError):
+                    os.kill(pid, 0)
+                removed.append(path)
+            return original_remove(path, *args, **kwargs)
+
+        with (
+            tempfile.TemporaryDirectory() as run_dir,
+            mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+            mock.patch.object(run_real_model.shutil, "rmtree", side_effect=remove),
+            mock.patch.object(run_real_model, "drive_hermes") as driver,
+            contextlib.redirect_stdout(io.StringIO()),
+            contextlib.redirect_stderr(io.StringIO()),
+        ):
+            result = run_real_model.run_scenario(
+                args, self.scenario(), Path(run_dir), report
             )
-        self.assertEqual(
-            [p.name for p in eval_env.root.glob("tracedecay-eval-*")], []
-        )
+        self.assertEqual(result, (False, False))
+        driver.assert_not_called()
+        self.assertTrue(removed)
+        self.assertFalse(eval_env.root.exists())
+        self.assertIn("command failed (1)", report["scenarios"][0]["error"])
+        self.assertNotIn("cleanup_error", report["scenarios"][0])
 
     def test_daemon_startup_failure_is_reported_and_cleaned(self):
         binary = self.make_bin()
@@ -348,7 +376,429 @@ def test_agent_failure_and_cancellation_write_error_entries(self):
                     entry["error"], str(exc) or "interrupted"
                 )
 
-    def test_daemon_cleanup_failure_preserves_report_and_removes_artifacts(self):
+    def test_daemon_cleanup_does_not_need_to_launch_a_helper(self):
+        eval_env = run_real_model.create_eval_environment("cleanup-no-helper")
+        self.addCleanup(eval_env.cleanup)
+        eval_env.start_daemon(self.make_bin(), ready_timeout=15)
+        daemon = eval_env.daemon
+        with mock.patch.object(
+            run_real_model.subprocess, "Popen", side_effect=OSError("cannot spawn helper")
+        ):
+            eval_env.cleanup()
+        self.assertIsNotNone(daemon.returncode)
+        self.assertIsNone(eval_env.daemon)
+        self.assertFalse(eval_env.root.exists())
+
+    def test_cleanup_failure_retains_original_error_when_artifact_stat_is_denied(self):
+        eval_env = run_real_model.create_eval_environment("cleanup-error")
+        self.addCleanup(eval_env.cleanup)
+        args = argparse.Namespace(tracedecay_bin=self.make_bin(), driver="hermes")
+        report = {"scenarios": []}
+        # Start before injecting stop failure so setup still uses real commands.
+        eval_env.start_daemon(args.tracedecay_bin, ready_timeout=15)
+        daemon = eval_env.daemon
+        original_stat = Path.stat
+        inaccessible = []
+
+        def stat(path, *args, **kwargs):
+            if path == eval_env.root or path == eval_env.socket_dir or path.parent == eval_env.root:
+                inaccessible.append(path)
+                raise PermissionError(errno.EACCES, "owned artifact stat denied", str(path))
+            return original_stat(path, *args, **kwargs)
+
+        with (
+            tempfile.TemporaryDirectory() as run_dir,
+            mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+            mock.patch.object(eval_env, "start_daemon"),
+            mock.patch.object(run_real_model, "seed_fixture"),
+            mock.patch.object(run_real_model, "drive_hermes", side_effect=RuntimeError("agent failed")),
+            mock.patch.object(run_real_model, "stop_owned_process", side_effect=OSError("stop failed")),
+            mock.patch.object(Path, "stat", autospec=True, side_effect=stat),
+        ):
+            result = run_real_model.run_scenario(args, self.scenario(), Path(run_dir), report)
+        self.assertEqual(result, (False, False))
+        self.assertEqual(report["scenarios"][0]["error"], "agent failed")
+        self.assertEqual(report["scenarios"][0]["cleanup_error"], "stop failed")
+        entry = report["scenarios"][0]
+        fixture = Path(entry["retained_fixture"])
+        self.assertEqual(entry["retained_profile"], str(eval_env.root))
+        self.assertEqual(entry["retained_socket_dir"], str(eval_env.socket_dir))
+        self.assertEqual(set(inaccessible), {fixture, eval_env.root, eval_env.socket_dir})
+        self.assertTrue(fixture.exists())
+        self.assertIs(eval_env.daemon, daemon)
+        self.assertIsNone(daemon.poll())
+        self.assertTrue(eval_env.root.exists())
+        eval_env.cleanup()
+        self.assertFalse(eval_env.root.exists())
+
+    def test_socket_cleanup_failure_reports_retained_directory_and_cleans_profile(self):
+        eval_env = run_real_model.create_eval_environment("socket-cleanup-error")
+        self.addCleanup(eval_env.cleanup)
+        binary = self.make_bin()
+        original_remove = run_real_model.shutil.rmtree
+        original_unlink = os.unlink
+        removed = []
+        fixture = None
+        daemon = None
+
+        def drive(_args, _scenario, path, _run_dir, _env):
+            nonlocal fixture, daemon
+            fixture = path
+            daemon = eval_env.daemon
+            return [{"turn_valid": True}]
+
+        def remove(path, *args, **kwargs):
+            path = Path(path)
+            if path in (eval_env.socket_dir, eval_env.root):
+                self.assertIsNone(eval_env.daemon)
+                self.assertIsNotNone(daemon.returncode)
+                removed.append(path)
+            return original_remove(path, *args, **kwargs)
+
+        def unlink(path, *args, **kwargs):
+            if Path(path).name == "daemon.sock":
+                raise PermissionError(errno.EACCES, "socket directory deletion failed", path)
+            return original_unlink(path, *args, **kwargs)
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with (
+                mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]),
+                mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=binary),
+                mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+                mock.patch.object(run_real_model, "drive_hermes", side_effect=drive),
+                mock.patch.object(run_real_model, "evaluate_assertions", return_value=[]),
+                mock.patch.object(run_real_model.shutil, "rmtree", side_effect=remove),
+                mock.patch.object(os, "unlink", side_effect=unlink),
+                contextlib.redirect_stdout(io.StringIO()),
+                contextlib.redirect_stderr(io.StringIO()),
+            ):
+                code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"])
+            report_path, = Path(tmp).glob("*/report.json")
+            report = json.loads(report_path.read_text())
+
+        self.assertEqual(code, 1)
+        self.assertEqual(report["status"], "fail")
+        self.assertEqual(removed, [eval_env.socket_dir, eval_env.root])
+        self.assertFalse(eval_env.root.exists())
+        self.assertFalse(fixture.exists())
+        self.assertTrue(eval_env.socket_dir.exists())
+        entry = report["scenarios"][0]
+        self.assertEqual(entry["status"], "error (cleanup failure)")
+        self.assertIn("socket directory deletion failed", entry["cleanup_error"])
+        self.assertEqual(entry["retained_socket_dir"], str(eval_env.socket_dir))
+        self.assertNotIn("retained_profile", entry)
+        self.assertNotIn("retained_fixture", entry)
+        self.assertNotIn("daemon_pid", entry)
+        eval_env.cleanup()
+        self.assertFalse(eval_env.socket_dir.exists())
+
+    def test_seed_failure_retains_fixture_until_daemon_stop_is_confirmed(self):
+        binary = self.make_bin()
+        eval_env = run_real_model.create_eval_environment("seed-cleanup-error")
+        self.addCleanup(eval_env.cleanup)
+        args = argparse.Namespace(tracedecay_bin=binary, driver="hermes")
+        scenario = self.scenario()
+        # Initialization and the fact write succeed before preload validation
+        # fails, so the daemon has already used the fixture being retained.
+        scenario["setup"]["facts"][0]["preload_searches"] = 1
+        report = {"scenarios": []}
+        fixture = None
+        original_seed = run_real_model.seed_fixture
+        original_stop = run_real_model.stop_owned_process
+
+        def seed(scenario, binary, env, path):
+            nonlocal fixture
+            fixture = path
+            return original_seed(scenario, binary, env, path)
+
+        def stop(process, grace):
+            if process is eval_env.daemon:
+                raise OSError("daemon stop failed")
+            return original_stop(process, grace)
+
+        with (
+            tempfile.TemporaryDirectory() as run_dir,
+            mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+            mock.patch.object(run_real_model, "seed_fixture", side_effect=seed),
+            mock.patch.object(run_real_model, "stop_owned_process", side_effect=stop),
+            mock.patch.object(run_real_model, "drive_hermes") as driver,
+            contextlib.redirect_stdout(io.StringIO()),
+            contextlib.redirect_stderr(io.StringIO()),
+        ):
+            result = run_real_model.run_scenario(args, scenario, Path(run_dir), report)
+
+        self.assertEqual(result, (False, False))
+        driver.assert_not_called()
+        entry = report["scenarios"][0]
+        self.assertIn("sets preload_searches without preload_query", entry["error"])
+        self.assertEqual(entry["cleanup_error"], "daemon stop failed")
+        self.assertIsNotNone(fixture)
+        self.assertTrue(fixture.exists(), "a live daemon must retain its seeded fixture")
+        self.assertEqual(entry["retained_fixture"], str(fixture))
+        self.assertEqual(entry["retained_profile"], str(eval_env.root))
+        self.assertEqual(entry["daemon_pid"], eval_env.daemon.pid)
+        self.assertIsNone(eval_env.daemon.poll())
+        self.assertEqual(
+            (fixture / "src/lib.rs").read_text(), "pub fn eval_fixture_marker() {}\n"
+        )
+        eval_env.cleanup()
+        self.assertFalse(eval_env.root.exists())
+        self.assertFalse(fixture.exists())
+
+    def test_signals_stop_agent_descendants_and_persist_cancelled_report(self):
+        # A real child and grandchild that ignore SIGTERM exercise forced
+        # process-group teardown rather than a mocked KeyboardInterrupt.
+        script = """
+import json, os, pathlib, signal, subprocess, sys, time
+signal.signal(signal.SIGTERM, signal.SIG_IGN)
+ready = pathlib.Path(sys.argv[1] + '.ready')
+child = subprocess.Popen([sys.executable, '-c',
+    'import pathlib,signal,sys,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); '
+    'pathlib.Path(sys.argv[1]).touch(); time.sleep(600)', str(ready)])
+while not ready.exists(): time.sleep(0.01)
+pathlib.Path(sys.argv[1]).write_text(json.dumps([os.getpid(), child.pid]))
+time.sleep(600)
+"""
+        for sig in (signal.SIGINT, signal.SIGTERM):
+            with self.subTest(signal=sig), tempfile.TemporaryDirectory() as tmp:
+                pidfile = Path(tmp) / "agent-pids.json"
+                eval_env = run_real_model.create_eval_environment("signal-cleanup")
+                self.addCleanup(eval_env.cleanup)
+                notified = threading.Event()
+
+                def interrupt_agent():
+                    deadline = time.monotonic() + 15
+                    while time.monotonic() < deadline:
+                        if pidfile.exists() and pidfile.stat().st_size:
+                            notified.set()
+                            os.kill(os.getpid(), sig)
+                            return
+                        time.sleep(0.01)
+
+                def drive(*_args):
+                    run_real_model.run([sys.executable, "-c", script, str(pidfile)], timeout=20)
+                    self.fail("the interrupted agent must not complete")
+
+                thread = threading.Thread(target=interrupt_agent, daemon=True)
+                previous_handlers = {s: signal.getsignal(s) for s in (signal.SIGINT, signal.SIGTERM)}
+                with (
+                    mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp) / "runs"),
+                    mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]),
+                    mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=self.make_bin()),
+                    mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+                    mock.patch.object(run_real_model, "drive_hermes", side_effect=drive),
+                    mock.patch.object(run_real_model, "PROCESS_STOP_GRACE_SECONDS", 0.2),
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    thread.start()
+                    code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"])
+                    thread.join(timeout=1)
+                self.assertTrue(notified.is_set())
+                self.assertEqual(code, 130)
+                report_path, = (Path(tmp) / "runs").glob("*/report.json")
+                report = json.loads(report_path.read_text())
+                self.assertEqual(report["status"], "cancelled")
+                self.assertEqual(report["scenarios"][0]["status"], "cancelled")
+                leader, child = json.loads(pidfile.read_text())
+                with self.assertRaises(ProcessLookupError):
+                    os.kill(leader, 0)
+                # Linux can retain an orphaned zombie until PID 1 reaps it;
+                # use the canonical runnable-member check for the descendant.
+                self.assertFalse(portable_process._group_alive(leader), child)
+                self.assertIsNone(eval_env.daemon)
+                self.assertFalse(eval_env.root.exists())
+                self.assertEqual(previous_handlers, {
+                    s: signal.getsignal(s) for s in (signal.SIGINT, signal.SIGTERM)
+                })
+
+    def test_cancelled_command_stop_failure_retains_its_files_and_primary_error(self):
+        script = """
+import os, pathlib, signal, sys, time
+signal.signal(signal.SIGTERM, signal.SIG_IGN)
+pathlib.Path(sys.argv[1]).write_text(str(os.getpid()))
+time.sleep(30)
+"""
+        # Exercise main before scenario allocation, fixture seeding, and an
+        # established agent turn with their respective process owners.
+        for phase in ("binary resolution", "fixture setup", "agent turn"):
+            with self.subTest(phase=phase), tempfile.TemporaryDirectory() as tmp:
+                pidfile = Path(tmp) / "command.pid"
+                eval_env = (
+                    None if phase == "binary resolution"
+                    else run_real_model.create_eval_environment("command-stop-failure")
+                )
+                fixture = None
+                command = None
+                stop_attempts = 0
+                notified = threading.Event()
+                finished = threading.Event()
+                original_stop = run_real_model.stop_owned_process
+                original_seed = run_real_model.seed_fixture
+
+                def interrupt_command():
+                    deadline = time.monotonic() + 15
+                    while time.monotonic() < deadline and not finished.wait(0.01):
+                        if pidfile.exists() and pidfile.stat().st_size:
+                            notified.set()
+                            os.kill(os.getpid(), signal.SIGTERM)
+                            return
+
+                def run_command():
+                    run_real_model.run(
+                        [sys.executable, "-c", script, str(pidfile)],
+                        cwd=fixture,
+                        timeout=20,
+                    )
+                    self.fail("the interrupted command must not complete")
+
+                def seed(scenario, binary, env, path):
+                    nonlocal fixture
+                    fixture = path
+                    if phase == "fixture setup":
+                        (path / "owned-input").write_text("must survive a failed stop")
+                        run_command()
+                    else:
+                        original_seed(scenario, binary, env, path)
+
+                def fail_command_stop(process, grace):
+                    nonlocal command, stop_attempts
+                    if str(pidfile) in process.args:
+                        command = process
+                        stop_attempts += 1
+                        raise OSError("command stop failed")
+                    return original_stop(process, grace)
+
+                thread = threading.Thread(target=interrupt_command, daemon=True)
+                try:
+                    with (
+                        mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp) / "runs"),
+                        mock.patch.object(run_real_model, "load_scenarios", return_value=[
+                            self.scenario(), {"id": "must-not-run-after-cancellation"},
+                        ]),
+                        mock.patch.object(
+                            run_real_model, "resolve_tracedecay_bin",
+                            side_effect=(
+                                (lambda _explicit: run_command()) if phase == "binary resolution"
+                                else (lambda _explicit: self.make_bin())
+                            ),
+                        ),
+                        mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env) as create_env,
+                        mock.patch.object(run_real_model, "seed_fixture", side_effect=seed),
+                        mock.patch.object(run_real_model, "drive_hermes", side_effect=lambda *_args: run_command()),
+                        mock.patch.object(run_real_model, "stop_owned_process", side_effect=fail_command_stop),
+                        mock.patch.object(run_real_model, "PROCESS_STOP_GRACE_SECONDS", 0.2),
+                        contextlib.redirect_stdout(io.StringIO()),
+                        contextlib.redirect_stderr(io.StringIO()),
+                    ):
+                        thread.start()
+                        code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"])
+                    finished.set()
+                    thread.join(timeout=1)
+                    self.assertTrue(notified.is_set())
+                    self.assertEqual(code, 130)
+                    self.assertEqual(stop_attempts, 2, "the enclosing owner must retry the command stop")
+                    self.assertIsNotNone(command)
+                    self.assertIsNone(command.poll())
+                    report_path, = (Path(tmp) / "runs").glob("*/report.json")
+                    report = json.loads(report_path.read_text())
+                    self.assertEqual(report["status"], "cancelled")
+                    if phase == "binary resolution":
+                        create_env.assert_not_called()
+                        self.assertEqual(report["scenarios"], [])
+                        self.assertEqual(report["phase"], phase)
+                        entry = report
+                        primary_error = entry["reason"]
+                    else:
+                        self.assertEqual(len(report["scenarios"]), 1)
+                        entry = report["scenarios"][0]
+                        self.assertEqual(entry["status"], "cancelled")
+                        primary_error = entry["error"]
+                        self.assertEqual(entry["retained_profile"], str(eval_env.root))
+                        self.assertEqual(entry["retained_fixture"], str(fixture))
+                        self.assertEqual(entry["daemon_pid"], eval_env.daemon.pid)
+                        self.assertTrue(eval_env.root.exists())
+                        self.assertTrue(fixture.exists())
+                        self.assertIsNone(eval_env.daemon.poll())
+                        if phase == "fixture setup":
+                            self.assertEqual(
+                                (fixture / "owned-input").read_text(),
+                                "must survive a failed stop",
+                            )
+                    self.assertIn("cancelled by signal", primary_error)
+                    self.assertEqual(entry["cleanup_error"], "command stop failed")
+                    self.assertEqual(entry["command_pid"], command.pid)
+                finally:
+                    finished.set()
+                    thread.join(timeout=1)
+                    if command is not None:
+                        original_stop(command, 0.2)
+                    if eval_env is not None:
+                        eval_env.cleanup()
+
+    def test_first_signal_during_command_stop_finishes_owned_group_cleanup(self):
+        script = """
+import json, os, pathlib, subprocess, sys, time
+ready = pathlib.Path(sys.argv[1] + '.ready')
+child = subprocess.Popen([sys.executable, '-c',
+    'import pathlib,signal,sys,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); '
+    'pathlib.Path(sys.argv[1]).touch(); time.sleep(30)', str(ready)],
+    stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+while not ready.exists(): time.sleep(0.01)
+pathlib.Path(sys.argv[1]).write_text(json.dumps([os.getpid(), child.pid]))
+"""
+        with tempfile.TemporaryDirectory() as tmp:
+            pidfile = Path(tmp) / "agent-pids.json"
+            eval_env = run_real_model.create_eval_environment("signal-during-stop")
+            self.addCleanup(eval_env.cleanup)
+            interrupted = False
+            original_signal_group = portable_process._signal_group
+
+            def interrupt_stop(pid, sig):
+                nonlocal interrupted
+                original_signal_group(pid, sig)
+                if not interrupted and sig == signal.SIGTERM and pidfile.exists():
+                    leader, _child = json.loads(pidfile.read_text())
+                    if pid == leader:
+                        interrupted = True
+                        os.kill(os.getpid(), signal.SIGTERM)
+
+            def drive(*_args):
+                run_real_model.run([sys.executable, "-c", script, str(pidfile)], timeout=20)
+                self.fail("the interrupted command stop must cancel the scenario")
+
+            try:
+                with (
+                    mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp) / "runs"),
+                    mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]),
+                    mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=self.make_bin()),
+                    mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+                    mock.patch.object(run_real_model, "drive_hermes", side_effect=drive),
+                    mock.patch.object(portable_process, "_signal_group", side_effect=interrupt_stop),
+                    mock.patch.object(run_real_model, "PROCESS_STOP_GRACE_SECONDS", 0.2),
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"])
+                self.assertTrue(interrupted)
+                self.assertEqual(code, 130)
+                leader, child = json.loads(pidfile.read_text())
+                self.assertFalse(portable_process._group_alive(leader), child)
+                self.assertIsNone(eval_env.daemon)
+                self.assertFalse(eval_env.root.exists())
+                report_path, = (Path(tmp) / "runs").glob("*/report.json")
+                entry = json.loads(report_path.read_text())["scenarios"][0]
+                self.assertEqual(entry["status"], "cancelled")
+                self.assertIn("cancelled by signal", entry["error"])
+                self.assertNotIn("cleanup_error", entry)
+            finally:
+                if pidfile.exists():
+                    leader, _child = json.loads(pidfile.read_text())
+                    portable_process._stop_group(leader, 0.2)
+
+
+    def test_daemon_cleanup_failure_retains_completed_results_and_owned_artifacts(self):
         binary = self.make_bin()
         for agent_failure, cleanup_failure in (
             (RuntimeError("agent failed"), OSError("cleanup failed")),
@@ -357,14 +807,21 @@ def test_daemon_cleanup_failure_preserves_report_and_removes_artifacts(self):
         ):
             with self.subTest(agent=agent_failure, cleanup=cleanup_failure):
                 environment = run_real_model.create_eval_environment("cleanup-failure")
+                self.addCleanup(environment.cleanup)
                 daemon_pid = None
-                outcomes = [{"name": "completed assertion", "passed": True}]
+                outcomes = [{
+                    "name": "completed assertion",
+                    "passed": True,
+                    "actual": 1,
+                    "op": "eq",
+                    "expected": 1,
+                }]
                 transcripts = [{"turn_valid": True}]
 
                 def stop_then_fail(env):
                     nonlocal daemon_pid
                     daemon_pid = env.daemon.pid
-                    run_real_model.stop_process(env.daemon, 5)
+                    run_real_model.stop_owned_process(env.daemon, 5)
                     raise cleanup_failure
 
                 with tempfile.TemporaryDirectory() as tmp:
@@ -386,18 +843,30 @@ def stop_then_fail(env):
                         )
                     report = json.loads(next(Path(tmp).glob("*/report.json")).read_text())
                 self.assertEqual(report["status"], "cancelled" if cancelled else "fail")
+                self.assertNotIn("reason", report)
                 entry = report["scenarios"][0]
                 if agent_failure is not None:
                     self.assertEqual(entry["error"], str(agent_failure))
-                    self.assertIn(str(cleanup_failure), entry["cleanup_error"])
                 else:
-                    self.assertIn(str(cleanup_failure), entry["error"])
+                    if cancelled:
+                        self.assertEqual(entry["error"], str(cleanup_failure))
+                    else:
+                        self.assertNotIn("error", entry)
                     self.assertEqual(entry["assertions"], outcomes)
                     self.assertEqual(entry["transcripts"], transcripts)
-                self.assertFalse(environment.root.exists())
-                self.assertFalse(environment.socket_dir.exists())
+                self.assertIn(str(cleanup_failure), entry["cleanup_error"])
+                # An interrupted/failed stop has not returned confirmation to
+                # its owner, even when this injected failure follows a reap.
+                self.assertTrue(environment.root.exists())
+                self.assertTrue(environment.socket_dir.exists())
+                self.assertEqual(entry["retained_profile"], str(environment.root))
+                self.assertTrue(Path(entry["retained_fixture"]).exists())
+                self.assertIsNotNone(environment.daemon)
                 with self.assertRaises(ProcessLookupError):
                     os.kill(daemon_pid, 0)
+                environment.cleanup()
+                self.assertFalse(environment.root.exists())
+                self.assertFalse(environment.socket_dir.exists())
 
 
 class ProcessCancellationTest(unittest.TestCase):
@@ -409,12 +878,33 @@ def test_stop_failure_still_kills_and_reaps_the_owned_process(self):
         try:
             with mock.patch.object(portable_process, "_stop_group", side_effect=OSError("stop failed")):
                 with self.assertRaisesRegex(OSError, "stop failed"):
-                    portable_process.stop_process(process, 1)
+                    portable_process.stop_owned_process(process, 1)
             self.assertIsNotNone(process.returncode)
             with self.assertRaises(ProcessLookupError):
                 os.kill(process.pid, 0)
         finally:
-            portable_process.stop_process(process, 1)
+            portable_process.stop_owned_process(process, 1)
+
+    def test_failed_forced_stop_preserves_the_stop_error_and_live_process_owner(self):
+        process = subprocess.Popen(
+            [sys.executable, "-c", "import time; time.sleep(600)"],
+            start_new_session=True,
+        )
+        stop_error = OSError("graceful stop failed")
+        force_error = PermissionError("forced stop denied")
+        try:
+            with (
+                mock.patch.object(portable_process, "_stop_group", side_effect=stop_error),
+                mock.patch.object(portable_process, "_signal_group", side_effect=force_error),
+                self.assertRaises(OSError) as raised,
+            ):
+                portable_process.stop_owned_process(process, 1)
+            self.assertIs(raised.exception, stop_error)
+            self.assertIs(raised.exception.__cause__, force_error)
+            self.assertIsNone(process.poll())
+            os.kill(process.pid, 0)
+        finally:
+            portable_process.stop_owned_process(process, 1)
 
     def test_signals_during_binary_resolution_and_turns_stop_owned_groups_and_write_reports(self):
         workload = """
@@ -484,7 +974,7 @@ def work(*args, **kwargs):
                         with self.assertRaises(ProcessLookupError):
                             os.kill(group, 0)
                     finally:
-                        run_real_model.stop_process(process, 1)
+                        run_real_model.stop_owned_process(process, 1)
                         if group is not None:
                             try:
                                 os.killpg(group, signal.SIGKILL)
@@ -513,6 +1003,103 @@ def test_timeout_stops_descendants_and_reaps_the_leader(self):
             ).returncode, 1)
 
 
+class ReportLifecycleTest(unittest.TestCase):
+    def test_command_preserves_captured_stderr_and_closes_both_pipes(self):
+        processes = []
+        original_popen = subprocess.Popen
+
+        def spawn(*args, **kwargs):
+            process = original_popen(*args, **kwargs)
+            processes.append(process)
+            return process
+
+        with mock.patch.object(run_real_model.subprocess, "Popen", side_effect=spawn):
+            result = run_real_model.run(
+                [sys.executable, "-c", "import sys; print('output'); print('detail', file=sys.stderr)"],
+                stderr=subprocess.PIPE,
+            )
+        self.assertEqual(result.stdout, "output\n")
+        self.assertEqual(result.stderr, "detail\n")
+        process, = processes
+        self.assertTrue(process.stdout.closed)
+        self.assertTrue(process.stderr.closed)
+
+    def test_command_cleanup_failure_keeps_the_original_command_error(self):
+        with mock.patch.object(
+            run_real_model, "stop_owned_process", side_effect=OSError("stop failed")
+        ):
+            with self.assertRaises(run_real_model.OwnedProcessCleanupError) as raised:
+                run_real_model.run([sys.executable, "-c", "print('original output'); exit(7)"])
+        error = raised.exception
+        self.assertIsInstance(error.original_error, run_real_model.EvalSetupError)
+        self.assertIn("command failed (7)", error.primary_detail())
+        self.assertIn("original output", error.primary_detail())
+        self.assertEqual(str(error.cleanup_error), "stop failed")
+        self.assertFalse(error.cancelled)
+        self.assertEqual(error.process.returncode, 7)
+        self.assertTrue(error.process.stdout.closed)
+
+    def test_resolution_failure_and_cancellation_always_write_report(self):
+        for error, expected_status, expected_code in (
+            (run_real_model.EvalSetupError("build failed"), "fail", 1),
+            (KeyboardInterrupt("build interrupted"), "cancelled", 130),
+        ):
+            with self.subTest(status=expected_status), tempfile.TemporaryDirectory() as tmp:
+                with (
+                    mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                    mock.patch.object(run_real_model, "load_scenarios", return_value=[{"id": "test"}]),
+                    mock.patch.object(run_real_model, "resolve_tracedecay_bin", side_effect=error),
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"])
+                self.assertEqual(code, expected_code)
+                report_path, = Path(tmp).glob("*/report.json")
+                report = json.loads(report_path.read_text())
+                self.assertEqual(report["status"], expected_status)
+                self.assertEqual(report["reason"], str(error))
+                self.assertEqual(report["phase"], "binary resolution")
+                self.assertEqual(report["scenarios"], [])
+
+    def test_cancellation_between_scenarios_and_during_report_write(self):
+        class InterruptedScenarios(list):
+            def __iter__(self):
+                yield self[0]
+                raise KeyboardInterrupt("between scenarios")
+
+        for during_write in (False, True):
+            with self.subTest(during_write=during_write), tempfile.TemporaryDirectory() as tmp:
+                scenarios = [{"id": "first"}] if during_write else InterruptedScenarios([{"id": "first"}])
+                original_write = Path.write_text
+                attempted = False
+
+                def write(path, contents, *args, **kwargs):
+                    nonlocal attempted
+                    if during_write and path.name == "report.json" and not attempted:
+                        attempted = True
+                        os.kill(os.getpid(), signal.SIGTERM)
+                    return original_write(path, contents, *args, **kwargs)
+
+                def run_scenario(_args, scenario, _run_dir, report):
+                    report["scenarios"].append({"id": scenario["id"], "status": "pass"})
+                    return True, False
+
+                with (
+                    mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                    mock.patch.object(run_real_model, "load_scenarios", return_value=scenarios),
+                    mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value="/unused"),
+                    mock.patch.object(run_real_model, "run_scenario", side_effect=run_scenario),
+                    mock.patch.object(Path, "write_text", new=write),
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    self.assertEqual(run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 130)
+                report_path, = Path(tmp).glob("*/report.json")
+                report = json.loads(report_path.read_text())
+                self.assertEqual(report["status"], "cancelled")
+                self.assertEqual(report["scenarios"], [{"id": "first", "status": "pass"}])
+
+
 class ExactFactStoreEvaluationTest(unittest.TestCase):
     def test_fixture_preload_requires_the_seeded_fact_in_search_results(self):
         scenario = {
@@ -541,15 +1128,18 @@ def test_fixture_preload_requires_the_seeded_fact_in_search_results(self):
                 ),
             ]
         )
-        with mock.patch.object(
-            run_real_model,
-            "run",
-            side_effect=lambda *args, **kwargs: next(responses),
+        with (
+            tempfile.TemporaryDirectory() as fixture,
+            mock.patch.object(
+                run_real_model,
+                "run",
+                side_effect=lambda *args, **kwargs: next(responses),
+            ),
         ):
             with self.assertRaisesRegex(
                 run_real_model.ExactFactStoreError, "did not return FactId"
             ):
-                run_real_model.build_fixture(scenario, "tracedecay", {})
+                run_real_model.seed_fixture(scenario, "tracedecay", {}, Path(fixture))
 
     def test_non_real_model_assertions_are_skipped_without_a_store_read(self):
         scenario = {
diff --git a/scripts/lib/portable_process.py b/scripts/lib/portable_process.py
index 29735bab8c..ff06cadf4c 100755
--- a/scripts/lib/portable_process.py
+++ b/scripts/lib/portable_process.py
@@ -165,23 +165,46 @@ def _stop_group(pid: int, grace_seconds: float) -> bool:
     return forced
 
 
-def stop_process(process: subprocess.Popen, grace_seconds: float) -> None:
-    """Stop an owned session and reap its leader, including on stop errors."""
+def _return_code(status: int) -> int:
+    return status if status >= 0 else 128 + abs(status)
+
+
+def stop_owned_process(process: subprocess.Popen, grace_seconds: float) -> None:
+    """Stop an owned session, including descendants, then reap its leader.
+
+    Callers must create the child with ``start_new_session=True``. This also
+    cleans children left behind by an already-exited leader, so a completed
+    command cannot leave work running against a fixture being removed.
+    """
+    stop_error = None
     try:
         _stop_group(process.pid, grace_seconds)
-    except BaseException:
-        _signal_group(process.pid, signal.SIGKILL)
-        raise
-    finally:
+    except Exception as exc:
+        stop_error = exc
+        # A failed graceful stop must still attempt to stop owned work. Keep
+        # the initial error if forced cleanup also fails; an interrupt remains
+        # a cancellation for the enclosing resource owner to finish.
+        try:
+            _signal_group(process.pid, signal.SIGKILL)
+        except Exception as cleanup_error:
+            raise stop_error from cleanup_error
+    try:
         try:
             process.wait(timeout=grace_seconds)
         except subprocess.TimeoutExpired:
-            _signal_group(process.pid, signal.SIGKILL)
+            process.kill()
             process.wait()
-
-
-def _return_code(status: int) -> int:
-    return status if status >= 0 else 128 + abs(status)
+        deadline = time.monotonic() + grace_seconds
+        while _group_alive(process.pid) and time.monotonic() < deadline:
+            time.sleep(min(0.05, max(0.0, deadline - time.monotonic())))
+        if _group_alive(process.pid):
+            raise RuntimeError(f"owned process group {process.pid} did not stop")
+    except Exception as cleanup_error:
+        if stop_error is not None:
+            raise stop_error from cleanup_error
+        raise
+    if stop_error is not None:
+        raise stop_error
 
 
 def command_exec_session(args: argparse.Namespace) -> int:

From fc718338807627c4b2c4c2c54e81f645a63379c2 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 01:00:13 -0700
Subject: [PATCH 06/12] fix(evals): clean profiles when environment creation is
 cancelled

---
 evals/memory/run_real_model.py      | 43 +++++++++++++++--------------
 evals/memory/test_run_real_model.py | 18 ++++++++++++
 2 files changed, 41 insertions(+), 20 deletions(-)

diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index 8ee38b6eca..fa6d830d32 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -242,33 +242,36 @@ def create_eval_environment(scenario_id):
     safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario_id)
     # Explicit ownership prevents an implicit temporary-directory finalizer
     # from deleting the profile if stopping an owned daemon fails.
-    root = Path(tempfile.mkdtemp(prefix=f"tracedecay-eval-env-{safe_id}-"))
-    home = root / "home"
-    data_dir = root / ".tracedecay"
-    global_db = data_dir / "global.db"
+    root = tempfile.mkdtemp(prefix=f"tracedecay-eval-env-{safe_id}-")
     try:
+        root = Path(root)
+        home = root / "home"
+        data_dir = root / ".tracedecay"
+        global_db = data_dir / "global.db"
         home.mkdir(parents=True, exist_ok=True)
         data_dir.mkdir(parents=True, exist_ok=True)
+
+        env = dict(os.environ)
+        env["HOME"] = str(home)
+        env["USERPROFILE"] = str(home)
+        env["XDG_CONFIG_HOME"] = str(home / ".config")
+        env["TRACEDECAY_DATA_DIR"] = str(data_dir)
+        env["TRACEDECAY_GLOBAL_DB"] = str(global_db)
+        env.pop("HERMES_HOME", None)
+        env.pop("HERMES_PROFILE", None)
+        return EvalEnvironment(
+            root=root,
+            home=home,
+            data_dir=data_dir,
+            global_db=global_db,
+            env=env,
+        )
     except BaseException:
+        # Ownership reaches the caller only after the whole environment has
+        # been constructed; cancellation during setup still belongs here.
         shutil.rmtree(root, ignore_errors=True)
         raise
 
-    env = dict(os.environ)
-    env["HOME"] = str(home)
-    env["USERPROFILE"] = str(home)
-    env["XDG_CONFIG_HOME"] = str(home / ".config")
-    env["TRACEDECAY_DATA_DIR"] = str(data_dir)
-    env["TRACEDECAY_GLOBAL_DB"] = str(global_db)
-    env.pop("HERMES_HOME", None)
-    env.pop("HERMES_PROFILE", None)
-    return EvalEnvironment(
-        root=root,
-        home=home,
-        data_dir=data_dir,
-        global_db=global_db,
-        env=env,
-    )
-
 
 def resolve_tracedecay_bin(explicit):
     """The shipping Bazel binary, built fresh, or the explicit one if it exists.
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index c98dcb70e0..b8f3c7ddc1 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -191,6 +191,24 @@ def test_eval_environment_pins_home_and_profile_storage_to_tempdir(self):
         self.assertTrue(env.data_dir.is_relative_to(env.root))
         self.assertNotEqual(env.global_db, Path.home() / ".tracedecay/global.db")
 
+    def test_cancellation_before_ownership_removes_the_profile(self):
+        for boundary in ("environment copy", "owner construction"):
+            with self.subTest(boundary=boundary), tempfile.TemporaryDirectory() as tmp:
+                cancelled = KeyboardInterrupt(f"cancelled during {boundary}")
+                interrupt = (
+                    mock.patch.object(os.environ, "keys", side_effect=cancelled)
+                    if boundary == "environment copy"
+                    else mock.patch.object(
+                        run_real_model, "EvalEnvironment", side_effect=cancelled
+                    )
+                )
+                with mock.patch.object(tempfile, "tempdir", tmp), interrupt as abort:
+                    with self.assertRaises(KeyboardInterrupt) as raised:
+                        run_real_model.create_eval_environment("cancelled-setup")
+                self.assertIs(raised.exception, cancelled)
+                abort.assert_called_once()
+                self.assertEqual(list(Path(tmp).iterdir()), [])
+
     def test_cleanup_removes_fixture_and_isolated_eval_store(self):
         eval_env = run_real_model.create_eval_environment("cleanup-store")
         self.addCleanup(eval_env.cleanup)

From d28a2a8381010cee2f8d32c6580ef1c8ddef6b04 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 01:05:26 -0700
Subject: [PATCH 07/12] fix(evals): preserve failed profile allocation cleanup

---
 evals/memory/run_real_model.py      |  38 ++++++--
 evals/memory/test_run_real_model.py | 132 ++++++++++++++++++++++++++--
 2 files changed, 159 insertions(+), 11 deletions(-)

diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index fa6d830d32..cbb039793e 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -238,6 +238,16 @@ def primary_detail(self):
         return str(error) or ("interrupted" if self.cancelled else type(error).__name__)
 
 
+class OwnedProfileCleanupError(EvalSetupError):
+    """Retain allocation failures before an environment can own its profile."""
+
+    def __init__(self, root, original_error, cleanup_error):
+        self.root = root
+        self.original_error = original_error
+        self.cleanup_error = cleanup_error
+        super().__init__(f"profile allocation cleanup failed for {root}: {cleanup_error}")
+
+
 def create_eval_environment(scenario_id):
     safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario_id)
     # Explicit ownership prevents an implicit temporary-directory finalizer
@@ -266,10 +276,15 @@ def create_eval_environment(scenario_id):
             global_db=global_db,
             env=env,
         )
-    except BaseException:
+    except BaseException as original_error:
         # Ownership reaches the caller only after the whole environment has
         # been constructed; cancellation during setup still belongs here.
-        shutil.rmtree(root, ignore_errors=True)
+        try:
+            shutil.rmtree(root)
+        except FileNotFoundError:
+            pass
+        except BaseException as cleanup_error:
+            raise OwnedProfileCleanupError(root, original_error, cleanup_error) from original_error
         raise
 
 
@@ -848,6 +863,8 @@ def run_scenario(args, scenario, run_dir, report):
     ok = False
     cancelled = False
     pending_command = None
+    unclaimed_profile = None
+    cleanup_error = None
 
     def cleanup():
         nonlocal pending_command
@@ -883,6 +900,13 @@ def cleanup():
         entry.pop("error")
         entry.update(status=status, assertions=outcomes, transcripts=transcripts)
         ok = not failed
+    except OwnedProfileCleanupError as exc:
+        unclaimed_profile = exc.root
+        cleanup_error = exc.cleanup_error
+        cancelled = isinstance(exc.original_error, KeyboardInterrupt)
+        entry["error"] = str(exc.original_error) or (
+            "interrupted" if cancelled else type(exc.original_error).__name__
+        )
     except OwnedProcessCleanupError as exc:
         pending_command = exc.process
         cancelled = exc.cancelled
@@ -907,20 +931,22 @@ def cleanup():
                 entry.setdefault("error", str(exc) or "interrupted")
                 cleanup()
         except BaseException as exc:
+            cleanup_error = exc
+        if cleanup_error is not None:
             ok = False
-            cancelled |= isinstance(exc, KeyboardInterrupt)
-            entry["cleanup_error"] = str(exc) or type(exc).__name__
+            cancelled |= isinstance(cleanup_error, KeyboardInterrupt)
+            entry["cleanup_error"] = str(cleanup_error) or type(cleanup_error).__name__
             if pending_command is not None:
                 entry["command_pid"] = pending_command.pid
             for label, path in (
                 ("retained_fixture", fixture),
-                ("retained_profile", eval_env.root if eval_env is not None else None),
+                ("retained_profile", eval_env.root if eval_env is not None else unclaimed_profile),
                 ("retained_socket_dir", eval_env.socket_dir if eval_env is not None else None),
             ):
                 if path is None:
                     continue
                 try:
-                    path.stat()
+                    os.stat(path)
                 except FileNotFoundError:
                     continue
                 except OSError:
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index b8f3c7ddc1..da0ced847f 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -302,6 +302,122 @@ def test_profile_allocation_failure_writes_error_report(self):
                 "transcripts": [],
             }])
 
+    def test_environment_construction_cancellation_removes_unowned_profile(self):
+        original_mkdtemp = tempfile.mkdtemp
+        for operation in ("dict", "EvalEnvironment"):
+            with self.subTest(operation=operation), tempfile.TemporaryDirectory() as tmp:
+                allocated = []
+
+                def allocate(*args, **kwargs):
+                    root = Path(original_mkdtemp(*args, **kwargs))
+                    allocated.append(root)
+                    self.addCleanup(run_real_model.shutil.rmtree, root, ignore_errors=True)
+                    return str(root)
+
+                cancellation = KeyboardInterrupt(f"cancelled during {operation}")
+                with (
+                    mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                    mock.patch.object(
+                        run_real_model, "load_scenarios", return_value=[
+                            self.scenario(), {**self.scenario(), "id": "must-not-start"}
+                        ]
+                    ),
+                    mock.patch.object(
+                        run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay"
+                    ),
+                    mock.patch.object(run_real_model.tempfile, "mkdtemp", side_effect=allocate),
+                    mock.patch.object(run_real_model, operation, side_effect=cancellation, create=True),
+                    mock.patch.object(run_real_model.subprocess, "Popen") as spawn,
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    self.assertEqual(
+                        run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 130
+                    )
+                spawn.assert_not_called()
+                reports = list(Path(tmp).glob("*/report.json"))
+                self.assertEqual(len(reports), 1)
+                report = json.loads(reports[0].read_text())
+                self.assertEqual(report["status"], "cancelled")
+                self.assertEqual(report["scenarios"], [{
+                    "id": "lifecycle",
+                    "contract": "stable",
+                    "status": "cancelled",
+                    "error": str(cancellation),
+                    "assertions": [],
+                    "transcripts": [],
+                }])
+                self.assertEqual(len(allocated), 1)
+                self.assertFalse(allocated[0].exists(), "unclaimed profile must be removed")
+
+    def test_allocation_cleanup_failure_preserves_cancellation_and_inaccessible_profile(self):
+        original_mkdtemp = tempfile.mkdtemp
+        original_remove = run_real_model.shutil.rmtree
+        original_stat = os.stat
+        for operation in ("Path", "EvalEnvironment"):
+            with self.subTest(operation=operation), tempfile.TemporaryDirectory() as tmp:
+                allocated = []
+
+                def allocate(*args, **kwargs):
+                    root = Path(original_mkdtemp(*args, **kwargs))
+                    allocated.append(root)
+                    self.addCleanup(original_remove, root, ignore_errors=True)
+                    return str(root)
+
+                def remove(path, *args, **kwargs):
+                    if allocated and Path(path) == allocated[0]:
+                        raise PermissionError("profile removal denied")
+                    return original_remove(path, *args, **kwargs)
+
+                def stat(path, *args, **kwargs):
+                    if (
+                        allocated
+                        and isinstance(path, (str, os.PathLike))
+                        and Path(path) == allocated[0]
+                    ):
+                        raise PermissionError("profile stat denied")
+                    return original_stat(path, *args, **kwargs)
+
+                cancellation = KeyboardInterrupt(f"cancelled during {operation}")
+                with (
+                    mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)),
+                    mock.patch.object(
+                        run_real_model, "load_scenarios", return_value=[
+                            self.scenario(), {**self.scenario(), "id": "must-not-start"}
+                        ]
+                    ),
+                    mock.patch.object(
+                        run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay"
+                    ),
+                    mock.patch.object(run_real_model.tempfile, "mkdtemp", side_effect=allocate),
+                    mock.patch.object(run_real_model, operation, side_effect=cancellation),
+                    mock.patch.object(run_real_model.shutil, "rmtree", side_effect=remove),
+                    mock.patch.object(os, "stat", side_effect=stat),
+                    mock.patch.object(run_real_model.subprocess, "Popen") as spawn,
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    self.assertEqual(
+                        run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 130
+                    )
+                spawn.assert_not_called()
+                self.assertEqual(len(allocated), 1)
+                self.assertTrue(allocated[0].exists())
+                reports = list(Path(tmp).glob("*/report.json"))
+                self.assertEqual(len(reports), 1)
+                report = json.loads(reports[0].read_text())
+                self.assertEqual(report["status"], "cancelled")
+                self.assertEqual(report["scenarios"], [{
+                    "id": "lifecycle",
+                    "contract": "stable",
+                    "status": "cancelled",
+                    "error": str(cancellation),
+                    "cleanup_error": "profile removal denied",
+                    "retained_profile": str(allocated[0]),
+                    "assertions": [],
+                    "transcripts": [],
+                }])
+
     def test_fresh_environment_inits_and_seeds_against_isolated_daemon(self):
         binary = self.make_bin()
         eval_env = run_real_model.create_eval_environment("lifecycle")
@@ -415,13 +531,19 @@ def test_cleanup_failure_retains_original_error_when_artifact_stat_is_denied(sel
         # Start before injecting stop failure so setup still uses real commands.
         eval_env.start_daemon(args.tracedecay_bin, ready_timeout=15)
         daemon = eval_env.daemon
-        original_stat = Path.stat
+        original_stat = os.stat
         inaccessible = []
 
         def stat(path, *args, **kwargs):
-            if path == eval_env.root or path == eval_env.socket_dir or path.parent == eval_env.root:
-                inaccessible.append(path)
-                raise PermissionError(errno.EACCES, "owned artifact stat denied", str(path))
+            if isinstance(path, (str, os.PathLike)):
+                artifact = Path(path)
+                if (
+                    artifact == eval_env.root
+                    or artifact == eval_env.socket_dir
+                    or artifact.parent == eval_env.root
+                ):
+                    inaccessible.append(artifact)
+                    raise PermissionError(errno.EACCES, "owned artifact stat denied", str(path))
             return original_stat(path, *args, **kwargs)
 
         with (
@@ -431,7 +553,7 @@ def stat(path, *args, **kwargs):
             mock.patch.object(run_real_model, "seed_fixture"),
             mock.patch.object(run_real_model, "drive_hermes", side_effect=RuntimeError("agent failed")),
             mock.patch.object(run_real_model, "stop_owned_process", side_effect=OSError("stop failed")),
-            mock.patch.object(Path, "stat", autospec=True, side_effect=stat),
+            mock.patch.object(os, "stat", side_effect=stat),
         ):
             result = run_real_model.run_scenario(args, self.scenario(), Path(run_dir), report)
         self.assertEqual(result, (False, False))

From b3101dda772d7e2793bd780db1ca4e7d73ddfbb3 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 04:59:13 -0700
Subject: [PATCH 08/12] docs(evals): clarify retained cleanup artifacts

---
 evals/memory/README.md | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/evals/memory/README.md b/evals/memory/README.md
index 8dcf129adb..a5d70af0ad 100644
--- a/evals/memory/README.md
+++ b/evals/memory/README.md
@@ -33,8 +33,9 @@ named Hermes profile or uses `HERMES_HOME` as TraceDecay storage.
 Each scenario also owns an isolated `tracedecay daemon run` (private socket,
 started via `scripts/lib/portable_process.py`) that must be ready before
 `init`. Setup, daemon, agent, and cancellation failures are written to the
-report as error entries, and the daemon, fixture, and profile are removed on
-every exit path.
+report as error entries. Cleanup is attempted on every exit path. If owned
+work cannot be stopped or deletion fails, retained paths and cleanup errors
+are recorded in the report.
 
 Run reports land under `evals/memory/runs/<timestamp>/` (gitignored).
 

From 233787935a94c20c665971bbdcc35b96a2b5e271 Mon Sep 17 00:00:00 2001
From: Zack Jackson <zackary.l.jackson@gmail.com>
Date: Wed, 21 Oct 2026 11:58:22 -0700
Subject: [PATCH 09/12] test(cli): follow profile authority in host daemon
 journeys

---
 .../sweep_outcomes.rs                         | 29 +++++++++++++++----
 1 file changed, 24 insertions(+), 5 deletions(-)

diff --git a/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs b/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs
index e447a37183..43d0713c74 100644
--- a/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs
+++ b/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs
@@ -7,6 +7,12 @@ use std::process::Output;
 
 use tracedecay_agent_hosts::agents::host_bundle::{HostComponentV1, HostKindV1};
 #[cfg(unix)]
+use tracedecay_daemon_identity::current_daemon_connection;
+#[cfg(unix)]
+use tracedecay_daemon_protocol::DaemonEndpoint;
+#[cfg(target_os = "linux")]
+use tracedecay_daemon_protocol::unix_socket_path_within_limit;
+#[cfg(unix)]
 use tracedecay_runtime_core::test_executable::write_executable_script;
 
 #[cfg(unix)]
@@ -387,13 +393,21 @@ impl ProfileDaemon {
             .stdout(std::process::Stdio::null())
             .stderr(std::process::Stdio::null());
         let daemon = Self(command.spawn().unwrap());
-        let socket = cli.profile.join("daemon.sock");
         let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
-        while std::os::unix::net::UnixStream::connect(&socket).is_err() {
+        // Deep worker paths can move the socket outside the profile. Follow
+        // the same profile-owned authority that production clients discover.
+        while !current_daemon_connection(&cli.profile).is_ok_and(|connection| {
+            match connection.endpoint() {
+                DaemonEndpoint::Unix(socket) => {
+                    std::os::unix::net::UnixStream::connect(socket).is_ok()
+                }
+                DaemonEndpoint::Loopback(address) => std::net::TcpStream::connect(address).is_ok(),
+            }
+        }) {
             assert!(
                 std::time::Instant::now() < deadline,
-                "the profile daemon never listened at {}",
-                socket.display()
+                "the registered daemon never listened for profile {}",
+                cli.profile.display()
             );
             std::thread::sleep(std::time::Duration::from_millis(25));
         }
@@ -864,7 +878,12 @@ fn doctor_observes_copilots_vscode_settings_without_failing_on_them() {
 #[cfg(target_os = "linux")]
 #[test]
 fn doctor_warns_on_detected_hosts_without_a_tracedecay_integration() {
-    let cli = IsolatedCli::new();
+    let mut cli = IsolatedCli::new();
+    cli.profile = cli.home.path().join("profile".repeat(20));
+    assert!(
+        !unix_socket_path_within_limit(&cli.profile.join("daemon.sock")),
+        "this journey must exercise production's short endpoint for a deep profile"
+    );
     install_cline(&cli);
     fs::create_dir_all(cli.home.path().join(".config/zed")).unwrap();
     fs::create_dir_all(cli.home.path().join(".gemini/antigravity")).unwrap();

From b7d9a27edc304b26b9a36796303cde9def0228c9 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 16:17:43 -0700
Subject: [PATCH 10/12] test(cli): preserve heavy refresh timeout diagnostics

---
 .../tests/refresh_memory_journey.rs           | 63 ++++++++++++++++++-
 crates/tracedecay/tests/common/mod.rs         | 23 ++++++-
 2 files changed, 83 insertions(+), 3 deletions(-)

diff --git a/crates/tracedecay-cli/tests/refresh_memory_journey.rs b/crates/tracedecay-cli/tests/refresh_memory_journey.rs
index 22f2ea026b..81bc33297a 100644
--- a/crates/tracedecay-cli/tests/refresh_memory_journey.rs
+++ b/crates/tracedecay-cli/tests/refresh_memory_journey.rs
@@ -17,6 +17,7 @@ use std::time::{Duration, Instant};
 
 use serde_json::{Value, json};
 use tempfile::TempDir;
+use tracedecay_runtime_core::logging::log_daemon_event;
 
 use common::fixture::{git_capture, git_run};
 use common::{
@@ -32,6 +33,17 @@ const FUNCTIONS_PER_FILE: usize = 160;
 const INDEX_TIMEOUT: Duration = Duration::from_secs(900);
 const STATUS_POLL: Duration = Duration::from_secs(5);
 
+fn report_phase(started: Instant, phase: &str, edit: usize) {
+    log_daemon_event(
+        "refresh_memory_journey_phase",
+        &[
+            ("phase", phase.to_owned()),
+            ("edit", edit.to_string()),
+            ("elapsed_ms", started.elapsed().as_millis().to_string()),
+        ],
+    );
+}
+
 /// Every function of every file changes with `edit`, so each refresh
 /// reparses the whole corpus.
 fn write_corpus(project: &Path, edit: usize) {
@@ -53,12 +65,25 @@ fn write_corpus(project: &Path, edit: usize) {
 }
 
 fn run_cli(home: &Path, project: &Path, args: &[&str]) -> Vec<u8> {
+    let started = Instant::now();
+    log_daemon_event(
+        "refresh_memory_journey_command_started",
+        &[("args", format!("{args:?}"))],
+    );
     let output = tracedecay_command_with_home(home)
         .args(args)
         .current_dir(project)
         .stdin(Stdio::null())
         .output()
         .unwrap_or_else(|error| panic!("tracedecay {args:?}: {error}"));
+    log_daemon_event(
+        "refresh_memory_journey_command_finished",
+        &[
+            ("args", format!("{args:?}")),
+            ("elapsed_ms", started.elapsed().as_millis().to_string()),
+            ("status", output.status.to_string()),
+        ],
+    );
     assert!(
         output.status.success(),
         "tracedecay {args:?} failed\nstdout:\n{}\nstderr:\n{}",
@@ -104,9 +129,19 @@ fn status(home: &Path, project: &Path) -> Value {
 /// Waits until the index serves a complete generation of `revision` and
 /// returns that generation's id.
 fn wait_for_current(home: &Path, project: &Path, revision: &str) -> String {
-    let deadline = Instant::now() + INDEX_TIMEOUT;
+    let started = Instant::now();
+    let deadline = started + INDEX_TIMEOUT;
     loop {
         let status = status(home, project);
+        log_daemon_event(
+            "refresh_memory_journey_status",
+            &[
+                ("expected_revision", revision.to_owned()),
+                ("elapsed_ms", started.elapsed().as_millis().to_string()),
+                ("freshness", status["code_index_freshness"].to_string()),
+                ("memory", status["memory"].to_string()),
+            ],
+        );
         let worktree = &status["code_index_freshness"]["worktree"];
         if status["code_index_freshness"]["status"] == "current"
             && worktree["source_revision"] == revision
@@ -156,6 +191,8 @@ fn assert_within_limit(home: &Path, project: &Path, pid: u32) {
 
 #[test]
 fn repeated_heavy_refreshes_under_the_limit_are_admitted() {
+    let started = Instant::now();
+    report_phase(started, "prepare_corpus", 0);
     let home = TempDir::new().expect("isolated home");
     let home = canonical_existing_path(home.path());
     let workspace = TempDir::new().expect("project root");
@@ -168,23 +205,43 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() {
     let cold_revision = git_capture(&project, &["rev-parse", "HEAD"]);
 
     let log = home.join("daemon.log");
+    report_phase(started, "start_daemon", 0);
     let daemon = spawn_tracedecay_daemon_logged(&home, &log, |command| {
         command
             .env(
                 "TRACEDECAY_RESIDENT_MEMORY_LIMIT_BYTES",
                 LIMIT_BYTES.to_string(),
             )
-            .env("RUST_LOG", "info");
+            .env("TRACEDECAY_TEST_DAEMON_LOG_STDERR", "1")
+            .env("TRACEDECAY_SPAN_TIMINGS", "1")
+            .env(
+                "RUST_LOG",
+                concat!(
+                    "info,",
+                    "tracedecay_code_index_runtime::code_index_scheduler::reconcile=trace,",
+                    "tracedecay_code_index_runtime::code_index_scheduler::registry=trace,",
+                    // Keep registry submodules at info: only its phase-level spans
+                    // should be timed, not per-file/page or serving-read work.
+                    "tracedecay_code_index_runtime::code_index_scheduler::registry::=info,",
+                    "tracedecay::daemon::project_open_admission=trace",
+                ),
+            );
     });
+    report_phase(started, "initialize_project", 0);
     initialize_tracedecay_cli_project(&home, &project);
+    report_phase(started, "wait_for_current", 0);
     let mut generation = wait_for_current(&home, &project, &cold_revision);
     assert_within_limit(&home, &project, daemon.id());
+    report_phase(started, "generation_current", 0);
 
     for edit in 1..=REFRESHES {
+        report_phase(started, "prepare_corpus", edit);
         write_corpus(&project, edit);
         git_run(&project, &["commit", "--quiet", "-am", "edit every file"]);
         let revision = git_capture(&project, &["rev-parse", "HEAD"]);
+        report_phase(started, "sync", edit);
         run_cli(&home, &project, &["sync"]);
+        report_phase(started, "wait_for_current", edit);
         let refreshed = wait_for_current(&home, &project, &revision);
         assert_ne!(
             refreshed, generation,
@@ -192,6 +249,7 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() {
         );
         generation = refreshed;
         assert_within_limit(&home, &project, daemon.id());
+        report_phase(started, "generation_current", edit);
     }
 
     let log = fs::read_to_string(&log).expect("daemon log");
@@ -206,4 +264,5 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() {
         "the daemon refused work it had room for:\n{log}"
     );
     drop(daemon);
+    report_phase(started, "complete", REFRESHES);
 }
diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs
index 827b125218..735d40db1b 100644
--- a/crates/tracedecay/tests/common/mod.rs
+++ b/crates/tracedecay/tests/common/mod.rs
@@ -693,6 +693,7 @@ fn drain_daemon_stderr(
     startup: String,
     mut stderr: BufReader<std::process::ChildStderr>,
     log: Option<PathBuf>,
+    mirror_stderr: bool,
 ) {
     std::thread::spawn(move || {
         if let Some(path) =
@@ -703,6 +704,18 @@ fn drain_daemon_stderr(
                 .open(path)
         {
             let _ = file.write_all(startup.as_bytes());
+            if mirror_stderr {
+                let mut line = Vec::new();
+                while stderr
+                    .read_until(b'\n', &mut line)
+                    .is_ok_and(|read| read != 0)
+                {
+                    let _ = file.write_all(&line);
+                    let _ = std::io::stderr().write_all(&line);
+                    line.clear();
+                }
+                return;
+            }
             let _ = std::io::copy(&mut stderr, &mut file);
             return;
         }
@@ -1165,6 +1178,8 @@ pub fn spawn_tracedecay_daemon_with(
 
 /// [`spawn_tracedecay_daemon_with`], streaming the daemon's stderr to `log`
 /// for assertions over its diagnostic events.
+/// Set `TRACEDECAY_TEST_DAEMON_LOG_STDERR=1` in `configure` to also stream the
+/// diagnostics to test stderr, preserving evidence when a runner kills a test.
 pub fn spawn_tracedecay_daemon_logged(
     home: &Path,
     log: &Path,
@@ -1276,6 +1291,9 @@ fn spawn_tracedecay_daemon_process(
         .stdout(Stdio::null())
         .stderr(Stdio::piped());
     configure(&mut command);
+    let mirror_stderr = command.get_envs().any(|(key, value)| {
+        key == "TRACEDECAY_TEST_DAEMON_LOG_STDERR" && value == Some(OsStr::new("1"))
+    });
     bind_to_test_process(&mut command);
     let child = command.spawn().expect("tracedecay daemon should start");
     let mut daemon = DaemonProcess::new(child);
@@ -1315,6 +1333,9 @@ fn spawn_tracedecay_daemon_process(
         let read = stderr
             .read_line(&mut startup)
             .expect("daemon stderr should be readable");
+        if mirror_stderr {
+            let _ = std::io::stderr().write_all(startup[line_start..].as_bytes());
+        }
         if read == 0 {
             let status = daemon
                 .child
@@ -1329,7 +1350,7 @@ fn spawn_tracedecay_daemon_process(
             break;
         }
     }
-    drain_daemon_stderr(startup, stderr, log);
+    drain_daemon_stderr(startup, stderr, log, mirror_stderr);
     daemon
 }
 

From 32bb5ff57bc4d800d5384104bcce21e72f14f0a7 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 17:30:45 -0700
Subject: [PATCH 11/12] test(memory): measure refreshes with routine daemon
 logging

---
 .../tracedecay-cli/tests/refresh_memory_journey.rs | 14 +-------------
 1 file changed, 1 insertion(+), 13 deletions(-)

diff --git a/crates/tracedecay-cli/tests/refresh_memory_journey.rs b/crates/tracedecay-cli/tests/refresh_memory_journey.rs
index 81bc33297a..19c902d0f9 100644
--- a/crates/tracedecay-cli/tests/refresh_memory_journey.rs
+++ b/crates/tracedecay-cli/tests/refresh_memory_journey.rs
@@ -213,19 +213,7 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() {
                 LIMIT_BYTES.to_string(),
             )
             .env("TRACEDECAY_TEST_DAEMON_LOG_STDERR", "1")
-            .env("TRACEDECAY_SPAN_TIMINGS", "1")
-            .env(
-                "RUST_LOG",
-                concat!(
-                    "info,",
-                    "tracedecay_code_index_runtime::code_index_scheduler::reconcile=trace,",
-                    "tracedecay_code_index_runtime::code_index_scheduler::registry=trace,",
-                    // Keep registry submodules at info: only its phase-level spans
-                    // should be timed, not per-file/page or serving-read work.
-                    "tracedecay_code_index_runtime::code_index_scheduler::registry::=info,",
-                    "tracedecay::daemon::project_open_admission=trace",
-                ),
-            );
+            .env("RUST_LOG", "info");
     });
     report_phase(started, "initialize_project", 0);
     initialize_tracedecay_cli_project(&home, &project);

From 91f30e7f42696ab8cd06f4b35cee73d219e2db55 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Wed, 21 Oct 2026 18:00:51 -0700
Subject: [PATCH 12/12] fix(evals): reject empty turns and repair macOS cleanup
 test

---
 evals/memory/run_real_model.py      |  2 +-
 evals/memory/test_run_real_model.py | 40 +++++++++++++++++++++++++++--
 2 files changed, 39 insertions(+), 3 deletions(-)

diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py
index cbb039793e..6afac4f6dd 100755
--- a/evals/memory/run_real_model.py
+++ b/evals/memory/run_real_model.py
@@ -894,7 +894,7 @@ def cleanup():
         status = "pass" if not failed else "fail"
         if failed and scenario.get("contract") == "pending-sibling":
             status = "fail (note: scenario contract is pending-sibling, see contract_notes)"
-        if not all(t["turn_valid"] for t in transcripts):
+        if not transcripts or not all(t["turn_valid"] for t in transcripts):
             status = "error (agent turn invalid, see transcript logs)"
             failed = failed or [{"name": "agent-turn", "passed": False}]
         entry.pop("error")
diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py
index da0ced847f..5a34f3d791 100644
--- a/evals/memory/test_run_real_model.py
+++ b/evals/memory/test_run_real_model.py
@@ -245,6 +245,33 @@ def test_cleanup_removes_fixture_and_isolated_eval_store(self):
 
 
 class IsolatedLifecycleTest(unittest.TestCase):
+    def test_no_agent_turns_cannot_pass_even_when_store_assertions_pass(self):
+        for driver in ("hermes", "cursor-agent"):
+            with self.subTest(driver=driver):
+                eval_env = run_real_model.create_eval_environment("empty-agent-turns")
+                self.addCleanup(eval_env.cleanup)
+                report = {"scenarios": []}
+                args = argparse.Namespace(driver=driver, tracedecay_bin="tracedecay")
+                with (
+                    mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env),
+                    mock.patch.object(eval_env, "start_daemon"),
+                    mock.patch.object(run_real_model, "seed_fixture"),
+                    mock.patch.object(run_real_model, "drive_hermes", return_value=[]),
+                    mock.patch.object(run_real_model, "drive_cursor_agent", return_value=[]),
+                    mock.patch.object(run_real_model, "evaluate_assertions", return_value=[
+                        {"name": "no-pollution", "passed": True, "actual": 0, "expected": 0, "op": "eq"},
+                    ]),
+                    contextlib.redirect_stdout(io.StringIO()),
+                    contextlib.redirect_stderr(io.StringIO()),
+                ):
+                    result = run_real_model.run_scenario(args, self.scenario(), Path("."), report)
+                self.assertEqual(result, (False, False))
+                entry = report["scenarios"][0]
+                self.assertEqual(entry["status"], "error (agent turn invalid, see transcript logs)")
+                self.assertEqual(entry["transcripts"], [])
+                self.assertTrue(entry["assertions"][0]["passed"])
+                self.assertFalse(eval_env.root.exists())
+
     def make_bin(self):
         tmp = tempfile.TemporaryDirectory()
         self.addCleanup(tmp.cleanup)
@@ -510,13 +537,22 @@ def test_agent_failure_and_cancellation_write_error_entries(self):
                     entry["error"], str(exc) or "interrupted"
                 )
 
-    def test_daemon_cleanup_does_not_need_to_launch_a_helper(self):
+    def test_daemon_cleanup_does_not_need_to_launch_a_python_helper(self):
         eval_env = run_real_model.create_eval_environment("cleanup-no-helper")
         self.addCleanup(eval_env.cleanup)
         eval_env.start_daemon(self.make_bin(), ready_timeout=15)
         daemon = eval_env.daemon
+        popen = subprocess.Popen
+
+        def process_state_only(command, *args, **kwargs):
+            # macOS has no /proc: canonical process-group teardown uses ps
+            # to distinguish exited zombies from descendants still running.
+            if command[0] == "ps":
+                return popen(command, *args, **kwargs)
+            raise OSError("cannot spawn cleanup helper")
+
         with mock.patch.object(
-            run_real_model.subprocess, "Popen", side_effect=OSError("cannot spawn helper")
+            run_real_model.subprocess, "Popen", side_effect=process_state_only
         ):
             eval_env.cleanup()
         self.assertIsNotNone(daemon.returncode)