From 36a7d67970dc7bffb2f3ac60ee5722549d496026 Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Sun, 20 Sep 2026 18:44:25 -0400
Subject: [PATCH 1/8] fix(evals): resolve terminal subagent failure crashing
 task

---
 evals/terminal-bench/README.md                |  8 ++-
 evals/terminal-bench/RELEASE_GATE.md          |  5 +-
 .../src/diffex_terminal_bench/trajectory.py   | 11 +++-
 evals/terminal-bench/tests/test_gate.py       | 20 +++++-
 evals/terminal-bench/tests/test_trajectory.py | 65 +++++++++++++++++++
 5 files changed, 103 insertions(+), 6 deletions(-)

diff --git a/evals/terminal-bench/README.md b/evals/terminal-bench/README.md
index 6d434a131..74b214956 100644
--- a/evals/terminal-bench/README.md
+++ b/evals/terminal-bench/README.md
@@ -409,7 +409,13 @@ Missing child sessions, missing observations, and incomplete child execution are
 Recoverable orphan child histories are embedded with an unresolved-link diagnostic.
 A child marked `interrupted` in the parent roster is a valid terminal state when a successful parent `agent_action` interrupt call establishes intent.
 An intentional stop can end during a model response or tool execution; it does not require an additional final assistant reply.
-Child provider errors and interruptions without that evidence remain invalid.
+A child marked `failed` in the final parent roster with a nonempty error is also a recorded terminal outcome.
+Its trajectory retains `execution_status: failed`, the scheduler's `execution_error`, and native response errors; it does not require a successful final assistant reply.
+The parent can finish normally despite that child failure, and the verifier independently determines the task reward.
+Missing child sessions, malformed records, unresolved tool calls, unproven interruptions, root execution failures, and cleanup failures still fail qualification.
+Missing usage telemetry remains explicitly incomplete and does not erase known usage.
+This policy runs in the local Python adapter and needs no new Diffex npm release.
+The staged runner requires a clean committed checkout and fresh output for the changed adapter identity; existing runs and retained reports are not rewritten.
 
 | Native Diffex usage | ATIF step metric | Harbor `AgentContext` |
 | --- | --- | --- |
diff --git a/evals/terminal-bench/RELEASE_GATE.md b/evals/terminal-bench/RELEASE_GATE.md
index 8e228173a..bf011a5df 100644
--- a/evals/terminal-bench/RELEASE_GATE.md
+++ b/evals/terminal-bench/RELEASE_GATE.md
@@ -260,11 +260,14 @@ Valid executions with missing usage telemetry retain their known usage sums as p
 Each checked trial records `usage_complete` and `usage_missing_fields` alongside its usage and verifier reward.
 An intentionally interrupted child is valid when its parent roster records `interrupted` and a successful parent interrupt call establishes intent.
 An aborted request without usage telemetry makes accounting partial; it does not invalidate the completed parent trial.
+A child whose final parent roster records `failed` with a nonempty error is a terminal failure that the parent can recover from.
+The exported child trajectory preserves its failed status and error, while missing telemetry remains partial and missing or corrupt evidence still fails validation.
+This local adapter policy requires no new Diffex release; commit the change and use a fresh output directory for subsequent staged runs.
 Verifier reward must agree with the retained reward file.
 A completed Diffex trial with valid evidence and reward zero passes integration checks and remains a zero-reward benchmark result.
 An oracle with reward zero fails the smoke.
 
-Timeouts, unrecovered provider or process errors, unexplained child interruptions, invalid usage values, missing logs, invalid ATIF, altered configuration, and cleanup failures retain their original evidence and mark the trial as failed.
+Timeouts, unrecovered root provider or process errors, unexplained child interruptions, invalid usage values, missing logs, invalid ATIF, altered configuration, and cleanup failures retain their original evidence and mark the trial as failed.
 The runner continues active peers and queued tasks after these failures, in both serial and concurrent stages.
 The runner does not convert them to successful executions or silently rerun them.
 After every selected task finishes, a stage with any failed trials reports failed qualification and exits nonzero with an evidence summary instead of a trial-error traceback.
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/trajectory.py b/evals/terminal-bench/src/diffex_terminal_bench/trajectory.py
index 875216d12..9774d1bcd 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/trajectory.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/trajectory.py
@@ -273,6 +273,7 @@ def export_trajectory(directory: Path, destination: Path, context: AgentContext,
     root_document = documents[root.id]
     linked = set()
     interrupted = set()
+    failed = set()
     for session in sessions:
         document = documents[session.id]
         children = []
@@ -285,6 +286,13 @@ def export_trajectory(directory: Path, destination: Path, context: AgentContext,
                 raise ValueError("cyclic or multiply-owned child session")
             linked.add(child_id)
             children.append(child_document)
+            # A recorded scheduler failure is terminal even when the child
+            # never produced a final answer. Keep the error in its trajectory;
+            # session integrity and tool-result checks still apply below.
+            error = child.get("error")
+            if child.get("status") == "failed" and isinstance(error, str) and error.strip():
+                failed.add(child_id)
+                child_document["extra"].update(execution_status="failed", execution_error=error)
             # Shutdown can also mark children interrupted. Accept a deliberate
             # stop only when the parent retained a successful interrupt call.
             if child.get("status") == "interrupted" and any(
@@ -329,7 +337,8 @@ def export_trajectory(directory: Path, destination: Path, context: AgentContext,
         if (session.id in interrupted and stop_reason == "error"
                 and assistants[-1]["extra"].get("error_message") == "This operation was aborted"):
             stop_reason = "aborted"
-        if stop_reason != "stop" and not (session.id in interrupted and stop_reason in (None, "aborted", "toolUse")):
+        if (stop_reason != "stop" and session.id not in failed
+                and not (session.id in interrupted and stop_reason in (None, "aborted", "toolUse"))):
             session.gaps.append("child_execution_incomplete")
     for session in sessions:
         last_assistant = next((step for step in reversed(session.steps)
diff --git a/evals/terminal-bench/tests/test_gate.py b/evals/terminal-bench/tests/test_gate.py
index d9802d394..23c697794 100644
--- a/evals/terminal-bench/tests/test_gate.py
+++ b/evals/terminal-bench/tests/test_gate.py
@@ -21,7 +21,7 @@
 from diffex_terminal_bench.outcome import parse_outcome
 from diffex_terminal_bench.trajectory import export_trajectory
 from test_outcome import retain_service, write_cleanup
-from test_trajectory import USAGE, assistant, entry, interrupted_child_records
+from test_trajectory import USAGE, assistant, entry, failed_child_records, interrupted_child_records
 
 
 class GateTests(unittest.TestCase):
@@ -90,7 +90,7 @@ def mutate(self, path, key, value):
         data[key] = value
         write_json(path, data)
 
-    def write_interrupted_child(self, **options):
+    def write_interrupted_child(self, *, records=None, **options):
         self.identity["policy"]["subagents"] = "enabled"
         for name in ("requested.json", "effective.json"):
             self.mutate(self.run / name, "subagents", "enabled")
@@ -99,7 +99,7 @@ def write_interrupted_child(self, **options):
             document = json.loads((self.directory / name).read_text())
             document["agent"]["kwargs"] = self.identity["policy"].copy()
             write_json(self.directory / name, document)
-        root, child = interrupted_child_records(**options)
+        root, child = records if records is not None else interrupted_child_records(**options)
         # Keep the final root response identical to the completed stdout stream.
         root[-1] = entry("one", assistant("response"))
         for name, records in (("root", root), ("child", child)):
@@ -122,6 +122,20 @@ def test_intentional_interruption_passes_with_partial_usage_and_zero_reward(self
         self.assertFalse(report["usage_complete"])
         self.assertEqual(report["usage_missing_fields"], ["cached_tokens", "completion_tokens", "cost_usd", "prompt_tokens"])
 
+    def test_recorded_child_failure_passes_with_zero_reward_and_partial_usage(self):
+        self.write_interrupted_child(records=failed_child_records())
+        report = validate_trial(self.directory, "task", self.identity)
+        self.assertEqual(report["rewards"], {"reward": 0})
+        self.assertEqual(report["usage"]["prompt_tokens"], 57)
+        self.assertFalse(report["usage_complete"])
+        self.mutate(self.directory / "sandbox-cleanup.json", "confirmed", False)
+        with self.assertRaisesRegex(ValueError, "Sandbox cleanup"):
+            validate_trial(self.directory, "task", self.identity)
+        self.mutate(self.directory / "sandbox-cleanup.json", "confirmed", True)
+        self.mutate(self.run / "process.json", "exit_code", 1)
+        with self.assertRaisesRegex(ValueError, "Native execution evidence"):
+            validate_trial(self.directory, "task", self.identity)
+
     def test_partial_accounting_still_rejects_corruption_and_execution_errors(self):
         cases = [{"status": "running"}, {"interrupt_error": True}, {"stop_reason": "error"},
                  {"usage": {**USAGE, "input": -1}}, {"usage": {**USAGE, "cost": {"total": "invalid"}}}]
diff --git a/evals/terminal-bench/tests/test_trajectory.py b/evals/terminal-bench/tests/test_trajectory.py
index 9e468274c..768c19881 100644
--- a/evals/terminal-bench/tests/test_trajectory.py
+++ b/evals/terminal-bench/tests/test_trajectory.py
@@ -46,6 +46,15 @@ def interrupted_child_records(*, status="interrupted", stop_reason="aborted", in
     return root, child
 
 
+def failed_child_records(*, error="Rate limit reached", stop_reason="error", usage=None):
+    root, child = interrupted_child_records(status="failed", stop_reason=stop_reason, usage=usage,
+                                            error_message=error)
+    # A failure is recorded by the scheduler without a parent interrupt call.
+    del root[2:4]
+    root[-2]["data"]["agents"][0]["error"] = error
+    return root, child
+
+
 class TrajectoryTests(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory()
@@ -154,6 +163,62 @@ def test_child_errors_and_unproven_interruptions_remain_invalid(self):
                 self.assertIn("child:child_execution_incomplete", accounting["gaps"])
                 self.assertIsNone(self.context.n_input_tokens)
 
+    def test_recorded_child_failure_retains_error_and_partial_usage(self):
+        root, child = failed_child_records()
+        self.session(root)
+        self.session(child, "child", "root")
+        document, accounting = self.export()
+        self.assertEqual(accounting["gaps"], [])
+        self.assertFalse(accounting["complete"])
+        self.assertEqual(accounting["totals"]["prompt_tokens"], 57)
+        self.assertEqual(self.context.n_input_tokens, 57)
+        failed = document["subagent_trajectories"][0]
+        self.assertEqual(failed["extra"]["execution_status"], "failed")
+        self.assertEqual(failed["extra"]["execution_error"], "Rate limit reached")
+        self.assertEqual(failed["steps"][-1]["extra"]["error_message"], "Rate limit reached")
+        self.assertNotIn("prompt_tokens", failed["steps"][-1]["metrics"])
+        self.assertFalse(failed["final_metrics"]["extra"]["accounting_complete"])
+        self.assertEqual(document["extra"]["execution_status"], "completed")
+
+    def test_recorded_runtime_failure_can_follow_a_completed_tool_call(self):
+        root, child = failed_child_records(error="Cannot continue from message role: assistant",
+                                          stop_reason="toolUse", usage=USAGE)
+        child[-1]["message"]["content"] = [{"type": "toolCall", "id": "read", "name": "read", "arguments": {}}]
+        child.append(entry("read-result", {"role": "toolResult", "toolCallId": "read", "content": "file", "isError": False}))
+        self.session(root)
+        self.session(child, "child", "root")
+        document, accounting = self.export()
+        self.assertEqual(accounting["gaps"], [])
+        self.assertTrue(accounting["complete"])
+        self.assertEqual(document["subagent_trajectories"][0]["extra"]["execution_error"],
+                         "Cannot continue from message role: assistant")
+        child.pop()
+        self.session(child, "child", "root")
+        _, accounting = self.export()
+        self.assertIn("child:missing_tool_result", accounting["gaps"])
+
+    def test_child_failure_requires_terminal_roster_evidence(self):
+        for status, error in (("running", "Rate limit reached"), ("queued", "Rate limit reached"),
+                              ("failed", None), ("failed", ""), ("failed", " "), ("failed", {})):
+            with self.subTest(status=status, error=error):
+                root, child = failed_child_records()
+                root[-2]["data"]["agents"][0].update(status=status, error=error)
+                self.session(root)
+                self.session(child, "child", "root")
+                _, accounting = self.export()
+                self.assertIn("child:child_execution_incomplete", accounting["gaps"])
+
+    def test_failed_child_does_not_hide_missing_or_corrupt_evidence(self):
+        root, child = failed_child_records()
+        self.session(root)
+        path = self.session(child, "child", "root")
+        path.write_text(path.read_text() + '{"type":\n')
+        _, accounting = self.export()
+        self.assertIn("child:invalid_record:child.jsonl", accounting["gaps"])
+        path.unlink()
+        _, accounting = self.export()
+        self.assertIn("root:missing_child_trajectory", accounting["gaps"])
+
     def test_intentional_abort_can_end_during_tool_call_generation(self):
         for call in ({"type": "toolCall", "id": "pending", "name": "bash", "arguments": {}},
                      {"type": "toolCall", "name": "bash"}):

From 37266d50e5804d787270d7fc3f802db586514675 Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Sun, 20 Sep 2026 23:55:02 -0400
Subject: [PATCH 2/8] fix(evals): preserve safe evidence symlinks

Accept confined relative file links so Hugging Face caches do not fail otherwise successful trials. Preserve link identity through inventories and Harbor exports while rejecting unsafe and cross-trial targets.
---
 evals/terminal-bench/RELEASE_GATE.md          |  7 ++
 .../src/diffex_terminal_bench/export_job.py   |  5 +-
 .../src/diffex_terminal_bench/gate.py         | 36 ++++++++-
 evals/terminal-bench/tests/test_export_job.py | 55 ++++++++++++-
 evals/terminal-bench/tests/test_gate.py       | 13 +--
 evals/terminal-bench/tests/test_inventory.py  | 79 +++++++++++++++++++
 6 files changed, 185 insertions(+), 10 deletions(-)
 create mode 100644 evals/terminal-bench/tests/test_inventory.py

diff --git a/evals/terminal-bench/RELEASE_GATE.md b/evals/terminal-bench/RELEASE_GATE.md
index bf011a5df..fd5100239 100644
--- a/evals/terminal-bench/RELEASE_GATE.md
+++ b/evals/terminal-bench/RELEASE_GATE.md
@@ -246,6 +246,13 @@ Trial directories contain Harbor `config.json`, `lock.json`, `result.json`, veri
 Diffex installation evidence includes the three CLI checks and verified npm/Node pins.
 The oracle does not install Diffex or call a model, so its Diffex installation, model usage, and ATIF checks are not applicable.
 
+Evidence inventories preserve relative symlinks that point directly to regular files within the same trial, including Hugging Face cache snapshot links.
+Link targets have a separate `symlink:` fingerprint; regular-file content hashes still detect changes to the referenced files.
+Absolute, escaping, dangling, chained, and directory symlinks are rejected, including paths that leave the trial and then reenter it.
+Exports preserve accepted links and validate containment even for failed trials that have no checked inventory.
+This evidence policy runs in the local Python eval harness and requires no new Diffex npm release.
+Start subsequent runs from the committed checkout with a fresh output directory; previously recorded failures remain unchanged.
+
 The gate re-parses native process/stream evidence instead of trusting the outcome summary alone.
 The execution policy includes `background_process_lifetime: environment` in the requested/effective configuration and benchmark identity.
 Pre-verifier evidence must establish that all agent execution stopped, with every retained task service backed by a durable environment ownership record.
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/export_job.py b/evals/terminal-bench/src/diffex_terminal_bench/export_job.py
index 8b9fda3e5..553b1e3c8 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/export_job.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/export_job.py
@@ -52,9 +52,12 @@ def export_job(source: Path, output: Path, *, allow_partial: bool = False) -> Pa
     locks = []
     common = None
     for directory in directories:
+        # Even failed/unchecked trials must have links confined to this trial,
+        # not merely to the encompassing stage scanned above.
+        trial_files = inventory(directory)
         if directory.name in checked:
             row = checked[directory.name]
-            if row["task"] != planned[directory.name] or inventory(directory) != row["files"]:
+            if row["task"] != planned[directory.name] or trial_files != row["files"]:
                 raise ValueError(f"{directory.name}: retained checked evidence changed")
         # Never silently omit a started trial, including failed or cancelled work.
         result = TrialResult.model_validate_json((directory / "result.json").read_text())
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/gate.py b/evals/terminal-bench/src/diffex_terminal_bench/gate.py
index c4342982b..dc820e373 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/gate.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/gate.py
@@ -4,8 +4,10 @@
 import hashlib
 import json
 import math
+import os
 from pathlib import Path
 import random
+import stat
 import tempfile
 
 from harbor.models.agent.context import AgentContext
@@ -79,12 +81,40 @@ def stage_tasks(stage: str, pilot: list[str], all_tasks: list[str], *, sample: d
 
 
 def inventory(directory: Path) -> dict[str, str]:
+    """Hash retained files and portable links without following link targets.
+
+    Cache snapshots may link directly to regular files inside the evidence root.
+    Bind the link text separately from file contents so retargeting, replacing a
+    link with a file, and editing its target all change the inventory.
+    """
+    if directory.is_symlink():
+        raise ValueError("Evidence root must not be a symlink")
+    directory = directory.resolve(strict=True)
     result = {}
     for path in sorted(directory.rglob("*")):
+        relative = path.relative_to(directory).as_posix()
         if path.is_symlink():
-            raise ValueError("Evidence must not contain symlinks")
-        if path.is_file():
-            result[path.relative_to(directory).as_posix()] = sha256(path)
+            link = os.readlink(path)
+            target = path.parent
+            if Path(link).is_absolute():
+                raise ValueError(f"Evidence symlinks must be relative: {relative}")
+            # Walk the original components, including '..', before resolving.
+            # Normalizing first could hide traversal through an external path
+            # or a directory symlink. Chained links are deliberately rejected.
+            try:
+                for part in Path(link).parts:
+                    if not stat.S_ISDIR(target.lstat().st_mode):
+                        raise ValueError(f"Evidence symlinks must not traverse other links or files: {relative}")
+                    target = target.parent if part == ".." else target / part
+                    if not target.is_relative_to(directory):
+                        raise ValueError(f"Evidence symlinks must stay inside their root: {relative}")
+                if not stat.S_ISREG(target.lstat().st_mode) or not path.is_file():
+                    raise ValueError(f"Evidence symlinks must point directly to regular files: {relative}")
+            except OSError as exc:
+                raise ValueError(f"Evidence symlinks must have existing regular-file targets: {relative}") from exc
+            result[relative] = "symlink:" + fingerprint({"target": link})
+        elif path.is_file():
+            result[relative] = sha256(path)
     return result
 
 
diff --git a/evals/terminal-bench/tests/test_export_job.py b/evals/terminal-bench/tests/test_export_job.py
index fbf211f87..89352811e 100644
--- a/evals/terminal-bench/tests/test_export_job.py
+++ b/evals/terminal-bench/tests/test_export_job.py
@@ -15,7 +15,7 @@
 from harbor.models.trial.result import AgentInfo, ExceptionInfo, ModelInfo, TrialResult
 from harbor.models.verifier.result import VerifierResult
 # These private APIs define the upload contract of the pinned Harbor release.
-from harbor.upload.uploader import _create_job_archive_file, _load_job_from_disk, _read_trial_lock_for_upload
+from harbor.upload.uploader import _create_job_archive_file, _create_trial_archive_file, _load_job_from_disk, _read_trial_lock_for_upload
 
 from diffex_terminal_bench.benchmark import PACKAGE, trial_config
 from diffex_terminal_bench.config import PINS
@@ -116,6 +116,59 @@ def test_failed_partial_export_includes_failure_and_unrun_count(self):
         self.assertIsNone(trials[6].agent_result)
         self.assertEqual(len(read_json(self.output / "export.json")["missing_trials"]), 3)
 
+    def test_cli_preserves_cache_links_through_harbor_archives(self):
+        directory = self.directories[0]
+        cache = directory / "agent/state/home/.cache/huggingface/hub/model"
+        (cache / "blobs").mkdir(parents=True)
+        (cache / "blobs/weights").write_bytes(b"model weights")
+        link = cache / "snapshots/revision/model.safetensors"
+        link.parent.mkdir(parents=True)
+        link.symlink_to("../../blobs/weights")
+        self.report["trials"][0]["files"] = inventory(directory)
+        write_json(self.source / "gate.json", self.report)
+        before = inventory(self.source)
+        command = subprocess.run([sys.executable, "-m", "diffex_terminal_bench.export_job", str(self.source),
+                                  "--output", str(self.output)], capture_output=True, text=True)
+        self.assertEqual(command.returncode, 0, command.stderr)
+        self.assertEqual(inventory(self.source), before)
+        relative = link.relative_to(directory)
+        exported = self.output / directory.name
+        self.assertTrue((exported / relative).is_symlink())
+        self.assertEqual((exported / relative).readlink(), Path("../../blobs/weights"))
+        for kind in ("trial", "job"):
+            with self.subTest(kind=kind):
+                archive = self.root / f"{kind}.tar.gz"
+                if kind == "trial":
+                    _create_trial_archive_file(exported, archive)
+                    prefix = Path()
+                else:
+                    _create_job_archive_file(self.output, archive)
+                    prefix = Path(self.output.name) / directory.name
+                restored = self.root / f"restored-{kind}"
+                with tarfile.open(archive) as tar:
+                    self.assertTrue(tar.getmember((prefix / relative).as_posix()).issym())
+                    tar.extractall(restored, filter="data")
+                restored_link = restored / prefix / relative
+                self.assertTrue(restored_link.is_symlink())
+                self.assertEqual(restored_link.read_bytes(), b"model weights")
+                self.assertEqual(inventory(restored / prefix / "agent"), inventory(exported / "agent"))
+
+        link.unlink()
+        link.symlink_to("../../blobs/./weights")
+        with self.assertRaisesRegex(ValueError, "checked evidence changed"):
+            export_job(self.source, self.root / "retargeted-export")
+
+    def test_failed_trial_cannot_link_to_another_trial(self):
+        directory = self.directories[0]
+        (directory / "agent/link").symlink_to(f"../../{self.directories[1].name}/agent/trajectory.json")
+        self.report.update(passed=False, trials=self.report["trials"][1:])
+        write_json(self.source / "gate.json", self.report)
+        # A stage-wide check alone would accept this link.
+        inventory(self.source)
+        with self.assertRaisesRegex(ValueError, "symlinks must stay inside"):
+            export_job(self.source, self.output, allow_partial=True)
+        self.assertFalse(self.output.exists())
+
     def test_sample_exports_only_selected_tasks_with_one_attempt_each(self):
         source = self.root / "sample-run/sample"
         selected = sample_tasks(list(self.identity["task_digests"]), 5, 42)
diff --git a/evals/terminal-bench/tests/test_gate.py b/evals/terminal-bench/tests/test_gate.py
index 23c697794..0d2eeb29a 100644
--- a/evals/terminal-bench/tests/test_gate.py
+++ b/evals/terminal-bench/tests/test_gate.py
@@ -6,6 +6,7 @@
 import io
 import json
 from pathlib import Path
+import shutil
 import tempfile
 import unittest
 from unittest.mock import AsyncMock, Mock, patch
@@ -189,6 +190,11 @@ def test_partial_accounting_and_harbor_totals_must_still_reconcile(self):
 
     def test_stage_continues_after_partial_usage_with_real_gate_validation(self):
         self.write_interrupted_child()
+        cache = self.run / "state/home/.cache/huggingface/hub/model"
+        (cache / "blobs").mkdir(parents=True)
+        (cache / "blobs/weights").write_bytes(b"weights")
+        (cache / "snapshots/revision").mkdir(parents=True)
+        (cache / "snapshots/revision/model.safetensors").symlink_to("../../blobs/weights")
         pilot = ["task", *[f"task-{index}" for index in range(2, 11)]]
         identity = {**self.identity, "task_digests": {name: "sha256:task" for name in pilot}}
         config = BenchmarkConfig.model_validate({**load_config(PACKAGE / "benchmark.toml").model_dump(),
@@ -199,11 +205,7 @@ def test_stage_continues_after_partial_usage_with_real_gate_validation(self):
         async def create(config):
             calls.append(config.trial_name)
             destination = config.trials_dir / config.trial_name
-            for path in self.directory.rglob("*"):
-                if path.is_file():
-                    target = destination / path.relative_to(self.directory)
-                    target.parent.mkdir(parents=True, exist_ok=True)
-                    target.write_bytes(path.read_bytes())
+            shutil.copytree(self.directory, destination, dirs_exist_ok=True, symlinks=True)
             return Mock(run=AsyncMock())
 
         manager = Mock(_cleanup=AsyncMock())
@@ -220,6 +222,7 @@ async def create(config):
         self.assertEqual(len(calls), 10)
         self.assertEqual([trial["task"] for trial in report["trials"]], pilot)
         for trial in report["trials"]:
+            self.assertTrue(any(value.startswith("symlink:") for value in trial["files"].values()))
             self.assertEqual(trial["evidence"]["rewards"], {"reward": 0})
             self.assertFalse(trial["evidence"]["usage_complete"])
             self.assertEqual(trial["evidence"]["usage"]["prompt_tokens"], 76)
diff --git a/evals/terminal-bench/tests/test_inventory.py b/evals/terminal-bench/tests/test_inventory.py
new file mode 100644
index 000000000..b2e916980
--- /dev/null
+++ b/evals/terminal-bench/tests/test_inventory.py
@@ -0,0 +1,79 @@
+"""Evidence inventory accepts portable cache links without weakening containment."""
+
+import os
+from pathlib import Path
+import tempfile
+import unittest
+
+from diffex_terminal_bench.gate import inventory
+
+
+class InventoryTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.parent = Path(temporary.name).resolve()
+        self.root = self.parent / "trial"
+        self.cache = self.root / "agent/state/home/.cache/huggingface/hub/model"
+        self.blob = self.cache / "blobs/weights"
+        self.blob.parent.mkdir(parents=True)
+        self.blob.write_bytes(b"model weights")
+        self.link = self.cache / "snapshots/revision/model.safetensors"
+        self.link.parent.mkdir(parents=True)
+
+    def test_cache_link_binds_target_text_type_and_contents(self):
+        self.link.symlink_to("../../blobs/weights")
+        before = inventory(self.root)
+        self.assertTrue(before[self.link.relative_to(self.root).as_posix()].startswith("symlink:"))
+        self.assertEqual(inventory(self.root), before)
+        self.assertEqual(os.readlink(self.link), "../../blobs/weights")
+        self.blob.write_bytes(b"changed weights")
+        self.assertNotEqual(inventory(self.root), before)
+        self.blob.write_bytes(b"model weights")
+        self.link.unlink()
+        self.link.symlink_to("../../blobs/./weights")
+        self.assertNotEqual(inventory(self.root), before)
+        self.link.unlink()
+        self.link.write_bytes(self.blob.read_bytes())
+        self.assertNotEqual(inventory(self.root), before)
+
+    def test_unsafe_link_targets_are_rejected(self):
+        outside = self.parent / "secret"
+        outside.write_text("must not be read")
+        cases = {
+            "absolute": str(self.blob),
+            "escape": "../secret",
+            "escape-and-reenter": "../trial/agent/state/home/.cache/huggingface/hub/model/blobs/weights",
+            "dangling": "missing",
+            "directory": "agent",
+            "chain": "alias",
+            "cycle": "link",
+            "file-traversal": "agent/state/home/.cache/huggingface/hub/model/blobs/weights/../weights",
+        }
+        # Test one link at a time so a different invalid link cannot mask it.
+        link = self.root / "link"
+        for name, target in cases.items():
+            with self.subTest(name=name):
+                if name == "chain":
+                    (self.root / "alias").symlink_to(self.blob.relative_to(self.root))
+                link.symlink_to(target)
+                with self.assertRaisesRegex(ValueError, "symlinks"):
+                    inventory(self.root)
+                link.unlink()
+                (self.root / "alias").unlink(missing_ok=True)
+
+    def test_directory_link_traversal_is_rejected(self):
+        (self.root / "z-alias").symlink_to(self.cache / "blobs", target_is_directory=True)
+        (self.root / "a-link").symlink_to("z-alias/weights")
+        with self.assertRaisesRegex(ValueError, "symlinks"):
+            inventory(self.root)
+
+    def test_linked_root_is_rejected(self):
+        alias = self.parent / "alias"
+        alias.symlink_to(self.root, target_is_directory=True)
+        with self.assertRaisesRegex(ValueError, "root must not be a symlink"):
+            inventory(alias)
+
+
+if __name__ == "__main__":
+    unittest.main()

From 75df3ff1157d524399fdb848bf1901096c4852dc Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Mon, 21 Sep 2026 12:44:38 -0400
Subject: [PATCH 3/8] fix(evals): disable benchmark subagents

---
 evals/terminal-bench/README.md               |  3 ++-
 evals/terminal-bench/RELEASE_GATE.md         |  6 +++---
 evals/terminal-bench/benchmark.toml          |  2 +-
 evals/terminal-bench/tests/test_execution.py | 15 +++++++++------
 4 files changed, 15 insertions(+), 11 deletions(-)

diff --git a/evals/terminal-bench/README.md b/evals/terminal-bench/README.md
index 74b214956..4c89b6f75 100644
--- a/evals/terminal-bench/README.md
+++ b/evals/terminal-bench/README.md
@@ -296,7 +296,8 @@ The pinned npm `0.3.1` release includes `dist/AGENTS.md`, which matches the adap
 The adapter retains that packaged context without adding a global duplicate.
 The effective system prompt must contain the release instructions, and their SHA-256 is recorded.
 
-The checked-in `benchmark.toml` and `execution.yaml` enable subagents, adding `spawn_agent`, `agent_action`, and `agent_status` to the explicit tool allowlist.
+The checked-in `benchmark.toml` disables subagents, omitting `spawn_agent`, `agent_action`, and `agent_status` from the explicit tool allowlist.
+The separate controlled smoke configuration in `execution.yaml` enables subagents and includes those tools.
 Diffex decides whether to call those tools; enabling them does not require a child on every task.
 Children share their trial's sandbox, and the pinned CLI owns child scheduling and its default concurrency limit.
 The staged runner's `-n/--concurrency` controls simultaneous trials and sandboxes separately.
diff --git a/evals/terminal-bench/RELEASE_GATE.md b/evals/terminal-bench/RELEASE_GATE.md
index fd5100239..fd2854058 100644
--- a/evals/terminal-bench/RELEASE_GATE.md
+++ b/evals/terminal-bench/RELEASE_GATE.md
@@ -28,7 +28,7 @@ The preparation command writes `results/release-gate/dataset.json` without creat
 
 Harbor 0.23.0 and Daytona SDK 0.210.0 are locked in the benchmark's private Python environment.
 Node 22.23.2, Diffex 0.3.1, npm integrity, registry, and Node archive checksums come from [pins.toml](src/diffex_terminal_bench/pins.toml).
-The default is `openai/gpt-5.6-luna`, `max` thinking, tools `read,bash,edit,write,update_plan`, enabled subagents, disabled evolution, and `context-only` project resources.
+The default is `openai/gpt-5.6-luna`, `max` thinking, tools `read,bash,edit,write,update_plan`, disabled subagents, disabled evolution, and `context-only` project resources.
 Enabled subagents add `spawn_agent`, `agent_action`, and `agent_status` to the model's tools; Diffex chooses whether to use them.
 Children execute inside the parent trial's sandbox and their sessions and usage remain in the trial evidence.
 Change and commit the configuration before starting a new qualification sequence.
@@ -137,8 +137,8 @@ open "$TB_OUTPUT/pilot"
 
 Keep the same terminal or restore the exact `TB_OUTPUT` value when continuing later.
 Do not change the checkout, configuration, or concurrency between steps 4-6.
-The recorded `gate.json` identity shows `concurrency: 3` and `policy.subagents: enabled`.
-Each Diffex run's `requested.json` and `effective.json` show the coordination tool allowlist; native sessions and `trajectory.json` show actual calls and child execution.
+The recorded `gate.json` identity shows `concurrency: 3` and `policy.subagents: disabled`.
+Each Diffex run's `requested.json` and `effective.json` show only the configured base tools; native sessions and `trajectory.json` show actual calls.
 
 ### Shared dataset cache and run index
 
diff --git a/evals/terminal-bench/benchmark.toml b/evals/terminal-bench/benchmark.toml
index b829ae23e..9876a5b80 100644
--- a/evals/terminal-bench/benchmark.toml
+++ b/evals/terminal-bench/benchmark.toml
@@ -28,6 +28,6 @@ task_count = 89
 [policy]
 thinking = "max"
 tools = "read,bash,edit,write,update_plan"
-subagents = "enabled"
+subagents = "disabled"
 project_resources = "context-only"
 background_process_lifetime = "environment"
diff --git a/evals/terminal-bench/tests/test_execution.py b/evals/terminal-bench/tests/test_execution.py
index a03fe9dd1..6118cda79 100644
--- a/evals/terminal-bench/tests/test_execution.py
+++ b/evals/terminal-bench/tests/test_execution.py
@@ -117,12 +117,16 @@ async def upload(source, target):
                                             "spawn_agent", "agent_action", "agent_status"])
         self.assertTrue(any("chown -R 1000" in call.kwargs["command"] for call in environment.exec.call_args_list))
 
-    async def test_checked_in_eval_policies_expose_subagent_tools(self):
+    async def test_checked_in_eval_policies_match_launch_tools(self):
         benchmark = load_config(PACKAGE / "benchmark.toml")
         staged = trial_config(benchmark, Path("/task"), Path("/runs/one"), "unique", oracle=False).agent
         controlled = JobConfig.model_validate(yaml.safe_load((PACKAGE / "execution.yaml").read_text())).agents[0]
-        for config in (staged, controlled):
-            with self.subTest(model=config.model_name):
+        base_tools = ["read", "bash", "edit", "write", "update_plan"]
+        for config, subagents, tools in (
+            (staged, "disabled", base_tools),
+            (controlled, "enabled", [*base_tools, "spawn_agent", "agent_action", "agent_status"]),
+        ):
+            with self.subTest(model=config.model_name, subagents=subagents):
                 environment = self.make_environment()
                 requested = []
 
@@ -134,9 +138,8 @@ async def upload(source, target):
                 agent = DiffexAgent(logs_dir=self.logs_dir, model_name=config.model_name, **config.kwargs,
                                     extra_env={"OPENAI_API_KEY": "test-only"})
                 await agent.run("Task", environment, AgentContext())
-                self.assertEqual(requested[0]["subagents"], "enabled")
-                self.assertEqual(requested[0]["tools"], ["read", "bash", "edit", "write", "update_plan",
-                                                       "spawn_agent", "agent_action", "agent_status"])
+                self.assertEqual(requested[0]["subagents"], subagents)
+                self.assertEqual(requested[0]["tools"], tools)
 
     async def test_missing_credentials_and_ambiguous_models_fail_before_exec(self):
         with patch.dict(os.environ, {}, clear=True):

From b022b8a942199a697ffdf1d65a003e82786167c0 Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Thu, 24 Sep 2026 09:06:15 -0400
Subject: [PATCH 4/8] fix(coding-agent): isolate shutdown evidence

Validate shutdown before bulk artifact transfer and retry transient evidence failures. Keep artifact errors verifier-eligible once shutdown is confirmed.
---
 .../src/diffex_terminal_bench/adapter.py      |  95 +++++++--
 .../src/diffex_terminal_bench/evidence.cjs    |  26 +++
 .../src/diffex_terminal_bench/outcome.py      |   4 +-
 .../src/diffex_terminal_bench/transport.py    |  33 +++
 evals/terminal-bench/tests/test_evidence.py   |  46 +++++
 evals/terminal-bench/tests/test_recovery.py   | 189 ++++++++++++++++--
 6 files changed, 363 insertions(+), 30 deletions(-)
 create mode 100644 evals/terminal-bench/src/diffex_terminal_bench/evidence.cjs
 create mode 100644 evals/terminal-bench/src/diffex_terminal_bench/transport.py
 create mode 100644 evals/terminal-bench/tests/test_evidence.py

diff --git a/evals/terminal-bench/src/diffex_terminal_bench/adapter.py b/evals/terminal-bench/src/diffex_terminal_bench/adapter.py
index 1cb80f0af..950ed2ec8 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/adapter.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/adapter.py
@@ -7,6 +7,7 @@
 import hashlib
 import json
 from pathlib import Path, PurePosixPath
+import shutil
 import tempfile
 import uuid
 
@@ -16,7 +17,9 @@
 
 from diffex_terminal_bench.config import PINS
 from diffex_terminal_bench.execution import ExecutionOptions, PROVIDER_ENV, SUBAGENT_TOOLS
+from diffex_terminal_bench.lifecycle import read_cleanup
 from diffex_terminal_bench.outcome import DiffexControlStopError, parse_outcome, require_completed
+from diffex_terminal_bench.transport import retry_transport
 from diffex_terminal_bench.trajectory import export_trajectory
 
 
@@ -116,7 +119,7 @@ async def run(
                 source = Path(temp, name)
                 source.write_text(content, encoding="utf-8")
                 await environment.upload_file(source, str(run_dir / name))
-            for name in ("run.cjs", "cleanup.cjs", "observe.mjs", "release-agents.md"):
+            for name in ("run.cjs", "cleanup.cjs", "evidence.cjs", "observe.mjs", "release-agents.md"):
                 with as_file(files(__package__).joinpath(name)) as source:
                     await environment.upload_file(source, str(run_dir / name))
         # Uploads may be root-owned even when the task selects a non-root user.
@@ -173,13 +176,18 @@ async def run(
             except Exception as exc:
                 # Export failure must not replace the original cancellation or
                 # provider exception, or let a completed run omit its ATIF file.
-                outcome.setdefault("recovery_errors", []).append({"phase": "trajectory_export", "error": type(exc).__name__})
+                outcome.setdefault("artifact_errors", []).append({"phase": "trajectory_export", "error": type(exc).__name__})
                 context.n_input_tokens = context.n_output_tokens = context.n_cache_tokens = context.cost_usd = None
                 context.model_usage = {}
                 context.metadata = {**(context.metadata or {}), "diffex_trajectory": {"validated": False}}
                 if outcome["status"] == "completed":
-                    outcome.update(status="infrastructure_failure", reason="trajectory_export_failed")
-            (self.logs_dir / run_dir.name / "outcome.json").write_text(json.dumps(outcome, indent=2) + "\n")
+                    outcome.update(status="artifact_collection_failure", reason="trajectory_export_failed")
+            try:
+                (directory / "outcome.json").write_text(json.dumps(outcome, indent=2) + "\n")
+            except OSError as exc:
+                outcome.setdefault("artifact_errors", []).append({"phase": "outcome_persistence", "error": type(exc).__name__})
+                if outcome["status"] == "completed":
+                    outcome.update(status="artifact_collection_failure", reason="outcome_persistence_failed")
             # Raise inside finally so even a pending Harbor deadline cannot
             # convert unsafe cleanup into a verifier-eligible agent timeout.
             if outcome.get("agent_execution_stopped") is not True:
@@ -201,20 +209,75 @@ async def _finish_run(self, environment: BaseEnvironment, run_dir: PurePosixPath
             ), timeout=15)
         except Exception as exc:
             failures.append({"phase": "cleanup", "error": type(exc).__name__})
-        try:
-            if environment.capabilities.mounted:
-                await asyncio.wait_for(environment.prepare_logs_for_host(), timeout=15)
-            else:
-                await asyncio.wait_for(environment.download_dir(str(run_dir), directory), timeout=30)
-        except Exception as exc:
-            failures.append({"phase": "log_collection", "error": type(exc).__name__})
-        outcome = parse_outcome(directory, execution["interruption"], execution["launcher_exit_code"])
+        # Keep the independently validated evidence outside the bulk download's
+        # destination so partial extraction cannot overwrite its stop decision.
+        with tempfile.TemporaryDirectory(prefix="diffex-shutdown-") as temp:
+            evidence_dir = Path(temp)
+            cleanup = None
+
+            async def fetch_evidence():
+                response = await self.exec_as_agent(
+                    environment,
+                    command=shlex.join(["/opt/diffex-terminal-bench/node/bin/node", str(run_dir / "evidence.cjs"), str(run_dir)]),
+                    timeout_sec=15,
+                )
+                evidence = json.loads(response.stdout)
+                if evidence is None:
+                    raise TimeoutError("Shutdown evidence is not yet published")
+                return evidence
+
+            try:
+                evidence = await retry_transport(fetch_evidence)
+                if not isinstance(evidence, dict):
+                    raise ValueError("Invalid shutdown evidence bundle")
+                for name, content in evidence.items():
+                    path = PurePosixPath(name)
+                    if (name not in ("cleanup.json", "stop-requested", "launcher.json", "cli.json", "process.json")
+                            and not (len(path.parts) == 2 and path.parts[0] == "ownership"
+                                     and path.name.endswith(".json") and name == path.as_posix())):
+                        raise ValueError("Unexpected shutdown evidence path")
+                    if not isinstance(content, str):
+                        raise ValueError("Invalid shutdown evidence content")
+                    target = evidence_dir / name
+                    target.parent.mkdir(parents=True, exist_ok=True)
+                    target.write_text(content)
+                cleanup = read_cleanup(evidence_dir)
+            except Exception as exc:
+                failures.append({"phase": "shutdown_evidence", "error": type(exc).__name__})
+
+            artifact_errors = []
+            try:
+                if environment.capabilities.mounted:
+                    await asyncio.wait_for(environment.prepare_logs_for_host(), timeout=15)
+                else:
+                    await asyncio.wait_for(environment.download_dir(str(run_dir), directory), timeout=30)
+            except Exception as exc:
+                artifact_errors.append({"phase": "log_collection", "error": type(exc).__name__})
+            # Retain this bundle even when invalid, for diagnosis. Never infer
+            # shutdown from files that arrive later with the bulk artifacts.
+            try:
+                shutil.copytree(evidence_dir, directory, dirs_exist_ok=True)
+            except OSError as exc:
+                artifact_errors.append({"phase": "evidence_persistence", "error": type(exc).__name__})
+            outcome = parse_outcome(directory, execution["interruption"], execution["launcher_exit_code"],
+                                    shutdown_directory=evidence_dir)
+            outcome.update(agent_execution_stopped=cleanup is not None, cleanup_confirmed=cleanup is not None)
+            if cleanup is not None:
+                outcome["retained_processes"] = cleanup["retained_processes"]
+            elif not execution["interruption"]:
+                outcome.update(status="cleanup_failure", reason="process_cleanup_unverified")
+            if artifact_errors:
+                outcome["artifact_errors"] = artifact_errors
+                # Partial downloads may still conclusively establish an agent
+                # failure. Keep that primary cause alongside artifact errors.
+                unknown = (outcome["status"] in ("completed", "incomplete")
+                           or (outcome["status"] == "process_failure" and not outcome.get("process"))
+                           or (outcome["status"] == "configuration_failure"
+                               and not all((directory / name).is_file() for name in ("requested.json", "effective.json"))))
+                if cleanup is not None and not execution["interruption"] and unknown:
+                    outcome.update(status="artifact_collection_failure", reason="log_collection_failed")
         if failures:
             outcome["recovery_errors"] = failures
-            if any(failure["phase"] == "cleanup" for failure in failures):
-                outcome.update(agent_execution_stopped=False, cleanup_confirmed=False)
-            if not execution["interruption"]:
-                outcome.update(status="infrastructure_failure", reason="recovery_failed")
         context.metadata = {**(context.metadata or {}), "diffex_outcome": outcome, "diffex_run_dir": str(run_dir)}
         # The caller writes the summary to host logs after shielded recovery.
         # Do not upload a second copy that could overwrite a later cancellation
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/evidence.cjs b/evals/terminal-bench/src/diffex_terminal_bench/evidence.cjs
new file mode 100644
index 000000000..61857ee6d
--- /dev/null
+++ b/evals/terminal-bench/src/diffex_terminal_bench/evidence.cjs
@@ -0,0 +1,26 @@
+// Transfer shutdown evidence without traversing logs, sessions, or agent state.
+const { readFileSync, readdirSync } = require("node:fs");
+const { join } = require("node:path");
+
+const directory = process.argv[2];
+const evidence = {};
+for (const name of ["cleanup.json", "stop-requested", "launcher.json", "cli.json", "process.json"]) {
+	try {
+		evidence[name] = readFileSync(join(directory, name), "utf8");
+	} catch (error) {
+		if (error.code !== "ENOENT") throw error;
+	}
+}
+// A lost cleanup RPC response can arrive before the remote process has
+// published its report. Absence is retryable; a negative report is not.
+if (!("cleanup.json" in evidence) || !("stop-requested" in evidence)) {
+	process.stdout.write("null");
+} else {
+	let ownership = [];
+	try { ownership = readdirSync(join(directory, "ownership")); }
+	catch (error) { if (error.code !== "ENOENT") throw error; }
+	for (const name of ownership.filter((name) => name.endsWith(".json")).sort()) {
+		evidence[`ownership/${name}`] = readFileSync(join(directory, "ownership", name), "utf8");
+	}
+	process.stdout.write(JSON.stringify(evidence));
+}
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/outcome.py b/evals/terminal-bench/src/diffex_terminal_bench/outcome.py
index 1c506f8f0..a4ccae89e 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/outcome.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/outcome.py
@@ -18,7 +18,7 @@ class DiffexControlStopError(RuntimeError):
 
 
 def parse_outcome(directory: Path, interruption: str | None = None,
-                  launcher_exit_code: int | None = None) -> dict:
+                  launcher_exit_code: int | None = None, *, shutdown_directory: Path | None = None) -> dict:
     result = {"status": "incomplete", "reason": "missing_evidence", "stream_complete": False,
               "assistant_stop_reason": None, "retry_attempts": 0, "recovered_retries": 0,
               "launcher_exit_code": launcher_exit_code, "interruption": interruption}
@@ -30,7 +30,7 @@ def parse_outcome(directory: Path, interruption: str | None = None,
     except (OSError, ValueError):
         process = {}
     try:
-        cleanup = read_cleanup(directory)
+        cleanup = read_cleanup(shutdown_directory if shutdown_directory is not None else directory)
         result.update(cleanup_confirmed=True, agent_execution_stopped=True,
                       retained_processes=cleanup["retained_processes"])
     except (OSError, ValueError, KeyError, TypeError, AttributeError):
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/transport.py b/evals/terminal-bench/src/diffex_terminal_bench/transport.py
new file mode 100644
index 000000000..d7b81c8d9
--- /dev/null
+++ b/evals/terminal-bench/src/diffex_terminal_bench/transport.py
@@ -0,0 +1,33 @@
+"""Bounded retries for idempotent Daytona evidence and volume requests."""
+
+import asyncio
+from collections.abc import Awaitable, Callable
+import random
+
+from daytona.common.errors import DaytonaConnectionError, DaytonaError, DaytonaTimeoutError
+
+
+async def retry_transport[T](operation: Callable[[], Awaitable[T]]) -> T:
+    # The outer bound includes backoff and all requests; each request is also
+    # bounded so a lost connection cannot consume the entire retry budget.
+    async with asyncio.timeout(90):
+        for attempt in range(3):
+            try:
+                async with asyncio.timeout(20):
+                    return await operation()
+            except Exception as exc:
+                transient = isinstance(exc, (TimeoutError, ConnectionError, DaytonaConnectionError, DaytonaTimeoutError))
+                if isinstance(exc, DaytonaError):
+                    transient |= exc.status_code in (408, 429, 500, 502, 503, 504)
+                if not transient or attempt == 2:
+                    raise
+                delay = 2 ** attempt + random.uniform(0, 0.25)
+                if isinstance(exc, DaytonaError):
+                    for key, value in exc.headers.items():
+                        if key.lower().startswith("retry-after"):
+                            try:
+                                delay = max(delay, float(value))
+                            except (ValueError, TypeError):
+                                pass
+                await asyncio.sleep(delay)
+    raise TimeoutError("Transport retry budget exhausted")
diff --git a/evals/terminal-bench/tests/test_evidence.py b/evals/terminal-bench/tests/test_evidence.py
new file mode 100644
index 000000000..80763719d
--- /dev/null
+++ b/evals/terminal-bench/tests/test_evidence.py
@@ -0,0 +1,46 @@
+"""Exercise the remote evidence collector without a provider or sandbox API."""
+
+from importlib.resources import files
+import json
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+from diffex_terminal_bench.lifecycle import read_cleanup
+from test_outcome import retain_service, write_evidence
+
+
+class EvidenceTests(unittest.TestCase):
+    def test_collector_only_transfers_shutdown_records(self):
+        with tempfile.TemporaryDirectory() as temp:
+            remote, local = Path(temp) / "remote", Path(temp) / "local"
+            write_evidence(remote)
+            report = retain_service(remote)
+            cache = remote / "state/home/.cache/packages"
+            cache.parent.mkdir(parents=True)
+            with cache.open("wb") as stream:
+                stream.truncate(3 * 1024 ** 3)
+            script = files("diffex_terminal_bench").joinpath("evidence.cjs")
+            result = subprocess.run(["node", str(script), str(remote)], capture_output=True,
+                                    text=True, check=True, timeout=5)
+            evidence = json.loads(result.stdout)
+            self.assertEqual(set(evidence), {"cleanup.json", "stop-requested", "launcher.json",
+                                             "cli.json", "process.json", "ownership/service.json"})
+            self.assertLess(len(result.stdout), 4096)
+            for name, content in evidence.items():
+                target = local / name
+                target.parent.mkdir(parents=True, exist_ok=True)
+                target.write_text(content)
+            self.assertEqual(read_cleanup(local), report)
+
+    def test_missing_required_evidence_is_pending(self):
+        with tempfile.TemporaryDirectory() as temp:
+            result = subprocess.run(["node", str(files("diffex_terminal_bench").joinpath("evidence.cjs")), temp],
+                                    capture_output=True, text=True, timeout=5)
+            self.assertEqual(result.returncode, 0)
+            self.assertIsNone(json.loads(result.stdout))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/evals/terminal-bench/tests/test_recovery.py b/evals/terminal-bench/tests/test_recovery.py
index 0d8563dd5..e955aae0a 100644
--- a/evals/terminal-bench/tests/test_recovery.py
+++ b/evals/terminal-bench/tests/test_recovery.py
@@ -8,13 +8,16 @@
 from unittest.mock import Mock, create_autospec
 from unittest.mock import patch
 
+from daytona.common.errors import DaytonaAuthenticationError, DaytonaConnectionError, DaytonaError
 from harbor.environments.base import BaseEnvironment, ExecResult
 from harbor.models.agent.context import AgentContext
+from harbor.models.task.config import TaskConfig
 from harbor.trial.single_step import SingleStepTrial
 
 from diffex_terminal_bench.adapter import DiffexAgent
 from diffex_terminal_bench.outcome import DiffexControlStopError, DiffexExecutionError
-from test_outcome import write_evidence
+from diffex_terminal_bench.transport import retry_transport
+from test_outcome import attempt, write_evidence
 from test_trajectory import assistant, entry
 
 
@@ -26,12 +29,20 @@ def setUp(self):
         self.environment = create_autospec(BaseEnvironment, instance=True)
         self.environment.default_user = None
         self.environment.capabilities.mounted = False
-        self.environment.exec.return_value = ExecResult(return_code=0, stdout="", stderr="")
+        remote = self.logs / "remote"
+        write_evidence(remote)
+        self.evidence = {path.name: path.read_text() for path in remote.iterdir()
+                         if path.name in ("cleanup.json", "stop-requested", "launcher.json", "cli.json", "process.json")}
+        self.environment.exec.side_effect = lambda **kwargs: self.exec_result(kwargs)
         self.environment.download_dir.side_effect = lambda source, target: write_evidence(Path(target))
         self.agent = DiffexAgent(logs_dir=self.logs, model_name="openai/gpt-5",
                                  extra_env={"OPENAI_API_KEY": "fixture-only"})
         self.context = AgentContext()
 
+    def exec_result(self, kwargs, return_code=0):
+        return ExecResult(return_code=return_code,
+                          stdout=json.dumps(self.evidence) if "evidence.cjs" in kwargs["command"] else "", stderr="")
+
     async def test_downloads_unmounted_logs_and_records_outcome(self):
         await self.agent.run("Task", self.environment, self.context)
         self.environment.download_dir.assert_awaited_once()
@@ -82,7 +93,7 @@ async def execute(**kwargs):
             elif "cleanup.cjs" in kwargs["command"]:
                 cleaning.set()
                 await finish_cleanup.wait()
-            return ExecResult(return_code=0, stdout="", stderr="")
+            return self.exec_result(kwargs)
 
         self.environment.exec.side_effect = execute
         task = asyncio.create_task(self.agent.run("Task", self.environment, self.context))
@@ -123,7 +134,7 @@ async def test_transport_failure_retains_logs_and_original_exception(self):
         async def execute(**kwargs):
             if "run.cjs" in kwargs["command"]:
                 raise ConnectionError("controlled transport loss")
-            return ExecResult(return_code=0, stdout="", stderr="")
+            return self.exec_result(kwargs)
 
         self.environment.exec.side_effect = execute
         with self.assertRaises(ConnectionError):
@@ -131,28 +142,66 @@ async def execute(**kwargs):
         self.environment.download_dir.assert_awaited_once()
         self.assertEqual(self.context.metadata["diffex_outcome"]["status"], "transport_failure")
 
-    async def test_cleanup_failure_still_collects_logs_and_fails_run(self):
+    async def test_cleanup_rpc_failure_with_valid_evidence_permits_run(self):
         async def execute(**kwargs):
-            return ExecResult(return_code=1 if "cleanup.cjs" in kwargs["command"] else 0, stdout="", stderr="")
+            if "cleanup.cjs" in kwargs["command"]:
+                raise ConnectionError("lost cleanup response")
+            return self.exec_result(kwargs)
 
         self.environment.exec.side_effect = execute
-        with self.assertRaises(DiffexControlStopError):
-            await self.agent.run("Task", self.environment, self.context)
+        await self.agent.run("Task", self.environment, self.context)
         self.environment.download_dir.assert_awaited_once()
-        self.assertEqual(self.context.metadata["diffex_outcome"]["status"], "infrastructure_failure")
+        self.assertEqual(self.context.metadata["diffex_outcome"]["status"], "completed")
+        self.assertEqual(self.context.metadata["diffex_outcome"]["recovery_errors"][0]["phase"], "cleanup")
 
     async def test_log_collection_failure_is_recorded_locally(self):
         self.environment.download_dir.side_effect = ConnectionError("fixture")
-        with self.assertRaises(DiffexControlStopError):
+        with self.assertRaises(DiffexExecutionError):
             await self.agent.run("Task", self.environment, self.context)
         record = json.loads(next(self.logs.glob("run-*/outcome.json")).read_text())
-        self.assertEqual(record["recovery_errors"][0]["phase"], "log_collection")
+        self.assertTrue(record["agent_execution_stopped"])
+        self.assertEqual(record["artifact_errors"][0]["phase"], "log_collection")
+
+    async def test_bulk_timeout_with_valid_shutdown_reaches_harbor_verifier(self):
+        remote = self.logs / "remote"
+        write_evidence(remote)
+        evidence = {str(path.relative_to(remote)): path.read_text() for path in remote.rglob("*")
+                    if path.is_file() and path.name in
+                    ("cleanup.json", "stop-requested", "launcher.json", "cli.json", "process.json")}
+
+        async def execute(**kwargs):
+            return ExecResult(return_code=0, stdout=json.dumps(evidence) if "evidence.cjs" in kwargs["command"] else "", stderr="")
+
+        self.environment.exec.side_effect = execute
+        self.environment.download_dir.side_effect = TimeoutError("bulk archive timed out")
+        trial = create_autospec(SingleStepTrial, instance=True)
+        trial.task = Mock(config=TaskConfig())
+        trial._agent_timeout_sec = 1
+        trial._load_trajectory = None
+        trial.result = self.context
+
+        async def agent_phase(**kwargs):
+            await self.agent.run("Task", self.environment, self.context)
+
+        async def run_agent():
+            await SingleStepTrial._run_agent(trial)
+
+        trial._run_agent_phase.side_effect = agent_phase
+        trial._run_agent.side_effect = run_agent
+        await SingleStepTrial._run(trial)
+        trial._run_verifier.assert_awaited_once()
+        outcome = self.context.metadata["diffex_outcome"]
+        self.assertTrue(outcome["agent_execution_stopped"])
+        self.assertEqual(outcome["status"], "artifact_collection_failure")
+        self.assertEqual(outcome["artifact_errors"][0]["phase"], "log_collection")
 
     async def test_deadline_with_failed_cleanup_blocks_real_harbor_verifier_path(self):
+        self.evidence["cleanup.json"] = json.dumps({"agent_execution_stopped": False})
+
         async def execute(**kwargs):
             if "run.cjs" in kwargs["command"]:
                 await asyncio.Event().wait()
-            return ExecResult(return_code=1 if "cleanup.cjs" in kwargs["command"] else 0, stdout="", stderr="")
+            return self.exec_result(kwargs, return_code=1 if "cleanup.cjs" in kwargs["command"] else 0)
 
         self.environment.exec.side_effect = execute
         trial = create_autospec(SingleStepTrial, instance=True)
@@ -171,6 +220,122 @@ async def agent_phase(**kwargs):
         trial._sync_agent_output.assert_awaited_once()
         self.assertEqual(self.context.metadata["diffex_outcome"]["status"], "cancelled")
 
+    async def test_evidence_is_validated_before_bulk_download(self):
+        def download(source, target):
+            self.assertTrue(any("evidence.cjs" in call.kwargs["command"]
+                                for call in self.environment.exec.call_args_list))
+            # A partial extraction must not invalidate the independent evidence.
+            Path(target, "cleanup.json").write_text("broken")
+            raise TimeoutError("bulk failed")
+
+        self.environment.download_dir.side_effect = download
+        with self.assertRaises(DiffexExecutionError):
+            await self.agent.run("Task", self.environment, self.context)
+        self.assertTrue(self.context.metadata["diffex_outcome"]["agent_execution_stopped"])
+        self.assertEqual(json.loads(next(self.logs.glob("run-*/cleanup.json")).read_text())["phase"], "pre-verifier")
+
+    async def test_invalid_evidence_blocks_even_if_bulk_later_contains_valid_cleanup(self):
+        valid = dict(self.evidence)
+        for name, content in (("cleanup.json", "{}"), ("launcher.json", "{}"),
+                              ("ownership/unexpected.json", "{}"), ("../escape", "bad")):
+            with self.subTest(name=name):
+                self.evidence = {**valid, name: content}
+                with self.assertRaises(DiffexControlStopError):
+                    await self.agent.run("Task", self.environment, self.context)
+                self.assertFalse(self.context.metadata["diffex_outcome"]["agent_execution_stopped"])
+
+    async def test_evidence_transient_failure_retries_before_collection(self):
+        attempts = 0
+
+        async def execute(**kwargs):
+            nonlocal attempts
+            if "evidence.cjs" in kwargs["command"]:
+                attempts += 1
+                self.environment.download_dir.assert_not_awaited()
+                if attempts < 3:
+                    raise DaytonaConnectionError("temporary")
+            return self.exec_result(kwargs)
+
+        self.environment.exec.side_effect = execute
+        with patch("diffex_terminal_bench.transport.asyncio.sleep") as sleep:
+            await self.agent.run("Task", self.environment, self.context)
+        self.assertEqual(attempts, 3)
+        self.assertEqual(sleep.await_count, 2)
+
+    async def test_evidence_retry_exhaustion_still_blocks_verification(self):
+        async def execute(**kwargs):
+            if "evidence.cjs" in kwargs["command"]:
+                raise TimeoutError("temporary")
+            return self.exec_result(kwargs)
+
+        self.environment.exec.side_effect = execute
+        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+            with self.assertRaises(DiffexControlStopError):
+                await self.agent.run("Task", self.environment, self.context)
+        self.assertEqual(sum("evidence.cjs" in call.kwargs["command"]
+                             for call in self.environment.exec.call_args_list), 3)
+
+    async def test_pending_cleanup_publication_is_retried(self):
+        attempts = 0
+
+        async def execute(**kwargs):
+            nonlocal attempts
+            if "cleanup.cjs" in kwargs["command"]:
+                raise ConnectionError("response lost while cleanup continues")
+            if "evidence.cjs" in kwargs["command"]:
+                attempts += 1
+                if attempts == 1:
+                    return ExecResult(return_code=0, stdout="null", stderr="")
+            return self.exec_result(kwargs)
+
+        self.environment.exec.side_effect = execute
+        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+            await self.agent.run("Task", self.environment, self.context)
+        self.assertEqual(attempts, 2)
+        self.assertTrue(self.context.metadata["diffex_outcome"]["agent_execution_stopped"])
+
+    async def test_local_artifact_write_failure_preserves_shutdown(self):
+        for phase in ("evidence_persistence", "outcome_persistence"):
+            with self.subTest(phase=phase):
+                original = Path.write_text
+
+                def write(path, *args, **kwargs):
+                    if path.name == "outcome.json":
+                        raise OSError("disk full")
+                    return original(path, *args, **kwargs)
+
+                target = "diffex_terminal_bench.adapter.shutil.copytree" if phase == "evidence_persistence" else "pathlib.Path.write_text"
+                with patch(target, side_effect=OSError("disk full")) if phase == "evidence_persistence" else patch(target, write):
+                    with self.assertRaises(DiffexExecutionError):
+                        await self.agent.run("Task", self.environment, self.context)
+                outcome = self.context.metadata["diffex_outcome"]
+                self.assertTrue(outcome["agent_execution_stopped"])
+                self.assertIn(phase, [error["phase"] for error in outcome["artifact_errors"]])
+
+    async def test_partial_download_preserves_conclusive_provider_failure(self):
+        def download(source, target):
+            write_evidence(Path(target), attempt("error", "503 provider failed"))
+            raise TimeoutError("later artifact transfer failed")
+
+        self.environment.download_dir.side_effect = download
+        with self.assertRaises(DiffexExecutionError):
+            await self.agent.run("Task", self.environment, self.context)
+        outcome = self.context.metadata["diffex_outcome"]
+        self.assertEqual(outcome["status"], "provider_failure")
+        self.assertTrue(outcome["agent_execution_stopped"])
+        self.assertEqual(outcome["artifact_errors"][0]["phase"], "log_collection")
+
+    async def test_transport_retry_respects_rate_limit_and_permanent_errors(self):
+        for error, count in ((DaytonaAuthenticationError("denied", status_code=401), 1),
+                             (DaytonaError("limited", status_code=429, headers={"Retry-After-authenticated": "5"}), 3)):
+            operation = create_autospec(self.agent._finish_run, side_effect=error)
+            with patch("diffex_terminal_bench.transport.asyncio.sleep") as sleep:
+                with self.assertRaises(type(error)):
+                    await retry_transport(lambda: operation(None, None, None, None))
+            self.assertEqual(operation.await_count, count)
+            if count == 3:
+                self.assertEqual([call.args[0] for call in sleep.call_args_list], [5, 5])
+
 
 if __name__ == "__main__":
     unittest.main()

From 7b3661d8ac59041602d0097929bb6c220d2cc2ee Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Thu, 24 Sep 2026 09:14:44 -0400
Subject: [PATCH 5/8] feat(coding-agent): persist benchmark logs

Keep task packages and caches on sandbox disk while retaining logs in isolated prefixes of one shared Daytona volume. Record every sandbox volume location for recovery after deletion.
---
 evals/terminal-bench/ARTIFACTS.md             |  41 +++++++
 evals/terminal-bench/README.md                |  22 ++--
 evals/terminal-bench/full.yaml                |   4 +-
 .../terminal-bench/scripts/smoke_execution.py |  11 +-
 .../diffex_terminal_bench/daytona_backend.py  |  50 +++++++-
 .../src/diffex_terminal_bench/run.cjs         |   9 +-
 evals/terminal-bench/tests/faux-provider.cjs  |  15 ++-
 evals/terminal-bench/tests/launcher.test.cjs  |  67 +++++++++++
 .../tests/test_daytona_backend.py             | 109 ++++++++++++++++++
 9 files changed, 309 insertions(+), 19 deletions(-)
 create mode 100644 evals/terminal-bench/ARTIFACTS.md
 create mode 100644 evals/terminal-bench/tests/launcher.test.cjs
 create mode 100644 evals/terminal-bench/tests/test_daytona_backend.py

diff --git a/evals/terminal-bench/ARTIFACTS.md b/evals/terminal-bench/ARTIFACTS.md
new file mode 100644
index 000000000..5f1884673
--- /dev/null
+++ b/evals/terminal-bench/ARTIFACTS.md
@@ -0,0 +1,41 @@
+# Artifact collection and shutdown verification
+
+`full.yaml` and the staged benchmark runner use `RecordedDaytonaEnvironment` with the pinned Harbor 0.23.0 and Daytona 0.210.0 APIs.
+The backend mounts the shared `diffex-terminal-bench-logs` volume at `/logs`, with a fresh `trials/<uuid>` subpath for each sandbox creation.
+Override `environment.kwargs.artifact_volume` to select a different project volume.
+The backend waits up to 120 seconds for volume readiness and records its name, ID, mount path, and subpath in the trial's `artifact-volume.json` before creating the sandbox.
+The file retains a `mounts` list with session and environment names, so separate verifiers and restarted environments do not replace the agent's recovery location.
+Harbor's Docker-in-Docker backend binds the corresponding `/logs` subdirectories from the sandbox host into the task container.
+
+Agent home, installed task packages, caches, and mutable harness state live in a private `/tmp/diffex-tb-state-*` directory on sandbox disk.
+They remain available to task services through verification and disappear with sandbox deletion.
+The exported `state.json` records these paths without linking the state tree into the logs.
+Logs, native sessions, transcripts, process identities, and ownership records live under `/logs/agent/run-*` on the volume.
+
+After requesting cleanup, the adapter retrieves only `cleanup.json`, `stop-requested`, available launcher/CLI identities, `process.json`, and ownership records through a separate small JSON response.
+It validates the report against the independent identity and ownership files before starting the bulk download.
+Retrieval makes at most three attempts, with a 20-second request limit and a 90-second total budget including exponential backoff, jitter, and any server `Retry-After` delay.
+A report that has not yet been published is retried; malformed or negative shutdown evidence is rejected.
+Bulk extraction cannot change the independently validated shutdown decision.
+
+Once shutdown is confirmed, bulk download, trajectory export, or local artifact persistence failures appear in `artifact_errors` and permit Harbor's verifier to run.
+Such a trial is still recorded as an artifact collection failure unless another conclusive execution failure or interruption already explains its outcome.
+Missing or invalid shutdown evidence raises `DiffexControlStopError` and continues to block verification.
+Release qualification still requires valid retained execution evidence; permitting verification does not qualify an incomplete artifact set.
+
+## Quotas and retention
+
+Daytona recommends a shared volume with isolated subpaths and allows at most 100 volumes per organization, so a 445-trial run consumes one project volume rather than 445 volumes.
+Volume data persists after sandbox deletion and does not count against sandbox disk quota.
+Volumes use FUSE and are slower than sandbox disk, which is why package installation and caches remain local.
+See the [Daytona volume documentation](https://www.daytona.io/docs/en/volumes/).
+
+The documented Tier 4 pool is 500 vCPUs, 1,000 GiB RAM, and 5,000 GiB sandbox storage, with 50,000 general API requests and 600 sandbox creations per minute.
+Keep job concurrency within the organization's remaining capacity, accounting for other running or stopped sandboxes and each task's resource requirements.
+See the [Daytona limits documentation](https://www.daytona.io/docs/en/limits/).
+This change does not increase job concurrency or sandbox resources.
+
+The shared volume is intentionally retained when a trial deletes its sandbox, including when local download fails.
+To recover artifacts, select the relevant session in `artifact-volume.json`, mount its volume ID and exact subpath on a temporary sandbox, and download `/logs`.
+Delete that recovery sandbox afterward.
+Remove only a trial's recorded subpath when its artifacts are no longer needed; do not delete the shared volume while any job uses it.
diff --git a/evals/terminal-bench/README.md b/evals/terminal-bench/README.md
index 4c89b6f75..46277fd38 100644
--- a/evals/terminal-bench/README.md
+++ b/evals/terminal-bench/README.md
@@ -30,6 +30,9 @@ See [Harbor's installed-agent documentation](https://www.harborframework.com/doc
 ## Full evaluation with additional sandbox storage
 
 The native full job uses [full.yaml](full.yaml): 89 tasks, five attempts each, three concurrent trials, and no automatic retries.
+It selects `RecordedDaytonaEnvironment` to retain logs and transcripts in one shared Daytona volume with isolated sandbox subpaths.
+For an existing command that uses `--env daytona` without this config, replace that option with `--env diffex_terminal_bench.daytona_backend:RecordedDaytonaEnvironment` to enable volume storage.
+See [artifact collection and retention](ARTIFACTS.md) for recovery records, retry bounds, and Tier 4 quota considerations.
 It requests 20 GiB of disk per sandbox so task files and Harbor's temporary log archive have more room than the failing 10 GiB allocation.
 This is an explicit resource override, not an evaluation with the official task resource limits unchanged.
 Daytona must allow the requested per-sandbox allocation and enough total storage for concurrent sandboxes.
@@ -105,6 +108,7 @@ These are installation checks, not Terminal-Bench task scores.
 
 ```bash
 uv run --locked python -m unittest discover -s tests -v
+node --test tests/launcher.test.cjs
 uv run --locked python scripts/smoke_execution.py --platform linux/arm64
 uv run --locked python scripts/smoke_execution.py --platform linux/amd64
 uv run --locked python scripts/smoke_outcomes.py --platform linux/arm64
@@ -286,7 +290,8 @@ The observer extension checks that the first user message contains exactly that
 It registers only a `message_start` observer, preserving compatibility with child agents.
 
 Each invocation creates an exclusive `run-<uuid>/` under Harbor's agent logs.
-Its fresh `state/` is the `DIFFEX_CODING_AGENT_DIR`, with a fresh home and global `settings.json` containing `{"evolution":{"enabled":false}}` before startup.
+Its `DIFFEX_CODING_AGENT_DIR` is a fresh private `/tmp/diffex-tb-state-*` directory on sandbox disk, with a fresh home and global `settings.json` containing `{"evolution":{"enabled":false}}` before startup.
+Packages and caches remain outside the exported logs and survive until sandbox teardown so retained task services can use them during verification.
 No prior settings, credentials, sessions, or evolved harness selections are copied into it.
 Startup network updates are disabled with the supported offline setting so the pinned model catalog remains stable.
 
@@ -317,9 +322,10 @@ Each run directory retains:
 - `ownership/*.json`: durable shell ownership records published by the CLI.
 - `stop-requested` and `cleanup.json`: the launch barrier, verified control shutdown, and explicitly retained environment-owned processes.
 - `launcher.stdout.log` and `launcher.stderr.log`: launcher diagnostics, including failures before CLI startup.
-- `outcome.json`: execution classification, stream completeness, final assistant stop reason, retry counts, interruption state, and recovery errors.
+- `outcome.json`: execution classification, stream completeness, final assistant stop reason, retry counts, interruption state, shutdown recovery errors, and separate artifact errors.
 - `accounting.json`: observed usage, nullable totals, per-model observations, and reasons accounting is incomplete.
-- `sessions/` and `state/`: native root/child history and isolated harness state.
+- `sessions/`: native root/child history.
+- `state.json`: sandbox-local harness state and home paths, without exporting their contents.
 
 Harbor collects the agent log tree after the run; avoid log filters that exclude these paths.
 `trajectory.json` sits directly under Harbor's agent log directory, with embedded child trajectories and optional `trajectory-media/` image files.
@@ -343,8 +349,8 @@ Tool errors alone do not fail execution: the assistant can inspect and recover f
 | Nonzero process, signal, or spawn error | `process_failure` | Raise `DiffexExecutionError` |
 | Missing, truncated, or unfinished native stream | `incomplete` | Raise `DiffexExecutionError` |
 | Effective configuration mismatch | `configuration_failure` | Raise `DiffexExecutionError` |
-| Unverified agent shutdown or missing shutdown evidence | `cleanup_failure` / `infrastructure_failure` | Raise `DiffexControlStopError`; block verification |
-| Failed evidence collection with verified shutdown | `infrastructure_failure` | Raise `DiffexExecutionError` |
+| Unverified agent shutdown or missing shutdown evidence | `cleanup_failure` | Raise `DiffexControlStopError`; block verification |
+| Failed artifact collection with verified shutdown | `artifact_collection_failure`, or preserve an existing failure | Record `artifact_errors`; permit verification |
 | Harbor task deadline cancels `run()` | `cancelled` | Preserve cancellation; Harbor raises `AgentTimeoutError` |
 | Environment transport timeout or failure | `timeout` / `transport_failure` | Preserve the original exception |
 
@@ -374,7 +380,9 @@ Harbor stops the shared environment after verification, including verifier failu
 The benchmark runner independently confirms sandbox deletion after every trial, including abandonment and transport failures.
 
 Recovery is shielded against repeated cancellation and runs before `run()` returns or re-raises the interruption.
-It allows up to 15 seconds for the cleanup transport and 15 seconds for mounted-log preparation or 30 seconds for remote log download.
+It allows up to 15 seconds for cleanup transport, then retrieves and validates only shutdown evidence with three bounded attempts within 90 seconds.
+After shutdown validation, it allows 15 seconds for mounted-log preparation or 30 seconds for bulk remote log download.
+Bulk download and local artifact persistence failures cannot invalidate independently confirmed shutdown.
 The final summary is written directly to host agent logs after recovery, so later remote log collection cannot overwrite a cancellation classification.
 These bounded recovery operations can extend the task deadline; they do not grant the agent additional task execution time.
 If the sandbox transport becomes unavailable, available host logs and recovery diagnostics remain, cleanup is not reported as successful, and Harbor's environment teardown remains necessary.
@@ -385,7 +393,7 @@ Avoid log filters that discard the retained evidence.
 The converter emits `ATIF-v1.8` using the models and validator shipped with pinned Harbor 0.23.0.
 See [Harbor's ATIF documentation](https://www.harborframework.com/docs/agents/trajectory-format) for the schema and validation command.
 It writes the final trajectory atomically after cleanup and evidence collection, before returning to Harbor.
-Export failure changes an otherwise completed execution to `infrastructure_failure`; it preserves an existing provider failure or cancellation.
+Export failure changes an otherwise completed execution to `artifact_collection_failure`; it preserves an existing provider failure or cancellation.
 An empty run with no recoverable steps retains accounting diagnostics without fabricating a trajectory.
 
 Native session message entries are authoritative, including history retained before compaction.
diff --git a/evals/terminal-bench/full.yaml b/evals/terminal-bench/full.yaml
index 5a80919d8..633cda18d 100644
--- a/evals/terminal-bench/full.yaml
+++ b/evals/terminal-bench/full.yaml
@@ -4,7 +4,9 @@ n_concurrent_trials: 3
 retry:
   max_retries: 0
 environment:
-  type: daytona
+  import_path: diffex_terminal_bench.daytona_backend:RecordedDaytonaEnvironment
+  kwargs:
+    artifact_volume: diffex-terminal-bench-logs
   override_storage_mb: 20480
 agents:
   - name: diffex_terminal_bench.adapter:DiffexAgent
diff --git a/evals/terminal-bench/scripts/smoke_execution.py b/evals/terminal-bench/scripts/smoke_execution.py
index 74a146169..a64e9c7ef 100644
--- a/evals/terminal-bench/scripts/smoke_execution.py
+++ b/evals/terminal-bench/scripts/smoke_execution.py
@@ -5,6 +5,7 @@
 import hashlib
 import json
 from pathlib import Path
+import shlex
 import subprocess
 import tempfile
 import tomllib
@@ -97,6 +98,9 @@ async def smoke(platform: str) -> None:
                 assert initial == {"state_entries": ["home", "settings.json"], "session_entries": [],
                                    "settings": {"evolution": {"enabled": False}}, "unrelated_credential_present": False}
                 assert (run_dir / "sessions/root.jsonl").stat().st_size > 0
+                state = json.loads((run_dir / "state.json").read_text())
+                assert state["state_dir"].startswith("/tmp/diffex-tb-state-")
+                assert not (run_dir / "state").exists()
                 events = [json.loads(line) for line in (run_dir / "stdout.jsonl").read_text().splitlines()]
                 assert not any(event.get("isError") for event in events)
                 if subagents == "enabled":
@@ -117,10 +121,9 @@ async def smoke(platform: str) -> None:
                 report["trajectories"].append({"run": run_dir.name, "validated": True, "context": context.model_dump()})
                 # Contaminate the previous trial after it completes, proving
                 # the next run never inherits auth, sessions, or harness state.
-                (run_dir / "state/auth.json").write_text('{"stale":true}')
-                (run_dir / "state/SYSTEM.md").write_text("STALE_HARNESS_SELECTION")
-                for selection in (run_dir / "state/harness-state/selections").glob("*.json"):
-                    selection.write_text('{"stale_harness_selection":true}')
+                await current.exec_as_agent(environment, command=shlex.join([
+                    "/opt/diffex-terminal-bench/node/bin/node", "/tmp/faux-provider.cjs", "contaminate", state["state_dir"],
+                ]))
             await agent.exec_as_root(environment, command="test ! -e /task/work/INJECTED && test \"$(cat cwd.txt)\" = /task/work && test \"$(cat shell.txt)\" = shell-ok && test \"$(cat result.txt)\" = file-ok && test \"$(cat child.txt)\" = child-ok")
             requests = [json.loads(line) for line in (trial.agent_dir / "fixture-requests.jsonl").read_text().splitlines()]
             assert requests and all(entry["credential_received"] for entry in requests)
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/daytona_backend.py b/evals/terminal-bench/src/diffex_terminal_bench/daytona_backend.py
index 8b404fb08..98b8a3186 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/daytona_backend.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/daytona_backend.py
@@ -3,15 +3,61 @@
 import asyncio
 from datetime import datetime, timezone
 from pathlib import Path
+from uuid import uuid4
 
-from daytona import AsyncDaytona, ListSandboxesQuery
-from daytona.common.errors import DaytonaNotFoundError
+from daytona import AsyncDaytona, CreateSandboxFromImageParams, CreateSandboxFromSnapshotParams, ListSandboxesQuery, VolumeMount
+from daytona.common.errors import DaytonaConflictError, DaytonaNotFoundError
 from harbor.environments.daytona import DaytonaEnvironment
 
 from diffex_terminal_bench.gate import read_json, write_json
+from diffex_terminal_bench.transport import retry_transport
 
 
 class RecordedDaytonaEnvironment(DaytonaEnvironment):
+    async def _create_sandbox(
+        self, params: CreateSandboxFromImageParams | CreateSandboxFromSnapshotParams,
+        daytona: AsyncDaytona | None = None,
+    ) -> None:
+        # Harbor 0.23.0 has no public hook for SDK volume mounts. This creation
+        # boundary covers both direct sandboxes and the DinD host (whose /logs
+        # paths Harbor bind-mounts into the task container).
+        if daytona is None:
+            if self._client_manager is None:
+                raise RuntimeError("Daytona client manager is not initialized")
+            daytona = await self._client_manager.get_client()
+        name = self._kwargs.get("artifact_volume", "diffex-terminal-bench-logs")
+        if not isinstance(name, str) or not name.strip():
+            raise ValueError("artifact_volume must be a nonempty volume name")
+
+        async def get_volume():
+            try:
+                return await daytona.volume.get(name, create=True)
+            except DaytonaConflictError:
+                # Another concurrent trial or runner created the shared volume.
+                return await daytona.volume.get(name)
+
+        async with asyncio.timeout(120):
+            volume = await retry_transport(get_volume)
+            while volume.state != "ready":
+                if volume.state not in ("pending_create", "creating"):
+                    raise RuntimeError(f"Artifact volume is not usable: {volume.state}")
+                await asyncio.sleep(1)
+                volume = await retry_transport(lambda: daytona.volume.get(name))
+
+        # One project volume, not one per trial: Daytona permits 100 volumes
+        # per organization. A random subpath isolates concurrent jobs/retries.
+        mount = VolumeMount(volume_id=volume.id, mount_path="/logs", subpath=f"trials/{uuid4().hex}")
+        params.volumes = [*(params.volumes or []), mount]
+        path = self.trial_paths.trial_dir / "artifact-volume.json"
+        record = read_json(path) if path.exists() else {"mounts": []}
+        record["mounts"].append({
+            "session_id": self.session_id, "environment_name": self.environment_name,
+            "volume_id": volume.id, "volume_name": name,
+            "mount_path": mount.mount_path, "subpath": mount.subpath,
+        })
+        write_json(path, record)
+        await super()._create_sandbox(params=params, daytona=daytona)
+
     async def start(self, force_build: bool) -> None:
         try:
             await super().start(force_build)
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/run.cjs b/evals/terminal-bench/src/diffex_terminal_bench/run.cjs
index 0421b896f..3c4c0e412 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/run.cjs
+++ b/evals/terminal-bench/src/diffex_terminal_bench/run.cjs
@@ -1,6 +1,6 @@
 // Launch the installed CLI with only the explicitly selected environment.
 const { spawn } = require("node:child_process");
-const { mkdirSync, readFileSync, writeFileSync, openSync, closeSync, existsSync, copyFileSync } = require("node:fs");
+const { mkdirSync, mkdtempSync, readFileSync, writeFileSync, openSync, closeSync, existsSync, copyFileSync } = require("node:fs");
 const { join } = require("node:path");
 const { identity } = require("./cleanup.cjs");
 
@@ -10,9 +10,10 @@ if (existsSync(join(runDir, "stop-requested"))) process.exit(1);
 const config = JSON.parse(readFileSync(join(runDir, "requested.json"), "utf8"));
 if (process.env.DIFFEX_TB_PROVIDER !== config.provider || process.env.DIFFEX_TB_MODEL !== config.model ||
 	process.env.DIFFEX_TB_THINKING !== config.thinking) throw new Error("Harbor environment differs from requested.json");
-const state = join(runDir, "state");
-// No recursive creation: reusing any prior state is an error.
-mkdirSync(state, { mode: 0o700 });
+// Keep packages, caches, credentials, and mutable harness state on sandbox
+// disk. Preserve them through verification for retained task services.
+const state = mkdtempSync("/tmp/diffex-tb-state-");
+writeFileSync(join(runDir, "state.json"), JSON.stringify({ state_dir: state, home: join(state, "home") }), { mode: 0o600 });
 mkdirSync(join(state, "home"), { mode: 0o700 });
 mkdirSync(join(runDir, "sessions"), { mode: 0o700 });
 writeFileSync(join(state, "settings.json"), JSON.stringify({ evolution: { enabled: false } }), { mode: 0o600 });
diff --git a/evals/terminal-bench/tests/faux-provider.cjs b/evals/terminal-bench/tests/faux-provider.cjs
index b077bd2f0..fd6bf679f 100644
--- a/evals/terminal-bench/tests/faux-provider.cjs
+++ b/evals/terminal-bench/tests/faux-provider.cjs
@@ -1,8 +1,21 @@
 // Controlled Linux smoke fixture. Never used by the benchmark adapter.
 const { createServer } = require("node:http");
-const { appendFileSync, readFileSync, readdirSync, writeFileSync } = require("node:fs");
+const { appendFileSync, existsSync, readFileSync, readdirSync, writeFileSync } = require("node:fs");
 const { join } = require("node:path");
 
+if (process.argv[2] === "contaminate") {
+	const state = process.argv[3];
+	writeFileSync(join(state, "auth.json"), '{"stale":true}');
+	writeFileSync(join(state, "SYSTEM.md"), "STALE_HARNESS_SELECTION");
+	const selections = join(state, "harness-state/selections");
+	if (existsSync(selections)) {
+		for (const name of readdirSync(selections).filter((name) => name.endsWith(".json"))) {
+			writeFileSync(join(selections, name), '{"stale_harness_selection":true}');
+		}
+	}
+	process.exit(0);
+}
+
 if (process.argv[2] === "prepare") {
 	const state = process.env.DIFFEX_CODING_AGENT_DIR;
 	const runDir = process.env.DIFFEX_TB_RUN_DIR;
diff --git a/evals/terminal-bench/tests/launcher.test.cjs b/evals/terminal-bench/tests/launcher.test.cjs
new file mode 100644
index 000000000..fd1ab719d
--- /dev/null
+++ b/evals/terminal-bench/tests/launcher.test.cjs
@@ -0,0 +1,67 @@
+const assert = require("node:assert/strict");
+const { spawnSync } = require("node:child_process");
+const { EventEmitter } = require("node:events");
+const { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } = require("node:fs");
+const { tmpdir } = require("node:os");
+const { join, resolve } = require("node:path");
+const { test } = require("node:test");
+const { runInNewContext } = require("node:vm");
+
+test("launcher preserves task packages outside exported logs and isolates each run", (t) => {
+	const root = mkdtempSync(join(tmpdir(), "diffex-launcher-test-"));
+	t.after(() => rmSync(root, { recursive: true, force: true }));
+	const states = [];
+	for (let index = 0; index < 2; index++) {
+		const logs = join(root, `run-${index}`);
+		mkdirSync(logs);
+		writeFileSync(join(logs, "requested.json"), JSON.stringify({
+			provider: "openai", model: "fixture", thinking: "high", tools: ["bash"],
+			credential_env_names: [], background_process_lifetime: "environment",
+		}));
+		writeFileSync(join(logs, "release-agents.md"), "fixture context");
+		const child = new EventEmitter();
+		child.pid = 12345;
+		const gate = new EventEmitter();
+		gate.end = (message) => assert.equal(message, "start\n");
+		child.stdio = [null, null, null, gate];
+		let childEnv;
+		runInNewContext(readFileSync(resolve(__dirname, "../src/diffex_terminal_bench/run.cjs"), "utf8"), {
+			require: (name) => {
+				if (name === "./cleanup.cjs") return { identity: () => "100" };
+				if (name === "node:child_process") return { spawn: (command, args, options) => {
+					childEnv = options.env;
+					assert.equal(command, "/bin/sh");
+					assert.ok(args.includes(join(logs, "ownership")));
+					return child;
+				} };
+				return require(name);
+			},
+			process: { argv: ["node", "run.cjs", logs], pid: 12344, cwd: () => root, on: () => {},
+				env: { PATH: process.env.PATH, DIFFEX_TB_PROVIDER: "openai", DIFFEX_TB_MODEL: "fixture", DIFFEX_TB_THINKING: "high" } },
+			setTimeout, clearTimeout,
+		});
+		const state = JSON.parse(readFileSync(join(logs, "state.json"), "utf8"));
+		t.after(() => rmSync(state.state_dir, { recursive: true, force: true }));
+		states.push(state.state_dir);
+		assert.equal(childEnv.HOME, state.home);
+		assert.equal(childEnv.DIFFEX_CODING_AGENT_DIR, state.state_dir);
+		assert.equal(childEnv.DIFFEX_SESSION_DIR, join(logs, "sessions"));
+		assert.ok(state.state_dir.startsWith("/tmp/diffex-tb-state-"));
+		assert.equal(existsSync(join(logs, "state")), false);
+		const cache = join(state.home, ".cache");
+		assert.equal(existsSync(cache), false);
+		mkdirSync(cache);
+		writeFileSync(join(cache, "package"), "task dependency");
+		const fixture = join(__dirname, "faux-provider.cjs");
+		const prepare = spawnSync(process.execPath, [fixture, "prepare"], { env: childEnv, encoding: "utf8" });
+		assert.equal(prepare.status, 0, prepare.stderr);
+		assert.equal(JSON.parse(readFileSync(join(logs, "fixture-start.json"), "utf8")).settings.evolution.enabled, false);
+		child.emit("close", 0, null);
+		assert.equal(readFileSync(join(cache, "package"), "utf8"), "task dependency");
+		assert.equal(JSON.parse(readFileSync(join(logs, "process.json"), "utf8")).exit_code, 0);
+		const contaminate = spawnSync(process.execPath, [fixture, "contaminate", state.state_dir], { encoding: "utf8" });
+		assert.equal(contaminate.status, 0, contaminate.stderr);
+		assert.deepEqual(JSON.parse(readFileSync(join(state.state_dir, "auth.json"), "utf8")), { stale: true });
+	}
+	assert.notEqual(states[0], states[1]);
+});
diff --git a/evals/terminal-bench/tests/test_daytona_backend.py b/evals/terminal-bench/tests/test_daytona_backend.py
new file mode 100644
index 000000000..c1afd9d54
--- /dev/null
+++ b/evals/terminal-bench/tests/test_daytona_backend.py
@@ -0,0 +1,109 @@
+"""Volume isolation at the pinned Harbor sandbox creation boundary."""
+
+import asyncio
+import json
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from daytona import CreateSandboxFromImageParams, CreateSandboxFromSnapshotParams, Image
+from daytona.common.errors import DaytonaConflictError, DaytonaError
+from harbor.environments.daytona import DaytonaEnvironment
+from harbor.models.job.config import JobConfig
+from harbor.models.task.config import EnvironmentConfig
+from harbor.models.trial.paths import TrialPaths
+import yaml
+
+from diffex_terminal_bench.daytona_backend import RecordedDaytonaEnvironment
+
+
+class VolumeTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        (self.root / "Dockerfile").write_text("FROM ubuntu:24.04\n")
+        self.environment = RecordedDaytonaEnvironment(
+            environment_dir=self.root, environment_name="fixture", session_id="fixture",
+            trial_paths=TrialPaths(self.root / "trial"), task_env_config=EnvironmentConfig(),
+        )
+        self.volume = SimpleNamespace(id="volume-id", state="ready")
+        self.client = SimpleNamespace(volume=SimpleNamespace(get=AsyncMock(return_value=self.volume)))
+        parent = patch.object(DaytonaEnvironment, "_create_sandbox", new_callable=AsyncMock)
+        self.create = parent.start()
+        self.addCleanup(parent.stop)
+
+    async def test_full_445_trials_share_one_volume_with_isolated_prefixes(self):
+        params = [CreateSandboxFromSnapshotParams(snapshot="fixture") for _ in range(445)]
+        # Exercise the same creation hook for every trial, with no live API.
+        for request in params:
+            await self.environment._create_sandbox(request, self.client)
+            await asyncio.sleep(0)
+        mounts = [request.volumes[0] for request in params]
+        self.assertEqual({mount.volume_id for mount in mounts}, {"volume-id"})
+        self.assertEqual({mount.mount_path for mount in mounts}, {"/logs"})
+        self.assertEqual(len({mount.subpath for mount in mounts}), 445)
+        self.assertTrue(all(mount.subpath.startswith("trials/") and ".." not in mount.subpath for mount in mounts))
+        self.assertEqual({call.args[0] for call in self.client.volume.get.call_args_list}, {"diffex-terminal-bench-logs"})
+        record = json.loads((self.root / "trial/artifact-volume.json").read_text())
+        self.assertEqual([entry["subpath"] for entry in record["mounts"]], [mount.subpath for mount in mounts])
+
+    async def test_separate_verifier_preserves_agent_recovery_location(self):
+        verifier = RecordedDaytonaEnvironment(
+            environment_dir=self.root, environment_name="verifier", session_id="verifier-session",
+            trial_paths=self.environment.trial_paths, task_env_config=EnvironmentConfig(),
+        )
+        for environment in (self.environment, verifier):
+            await environment._create_sandbox(CreateSandboxFromSnapshotParams(snapshot="fixture"), self.client)
+        entries = json.loads((self.root / "trial/artifact-volume.json").read_text())["mounts"]
+        self.assertEqual([entry["session_id"] for entry in entries], ["fixture", "verifier-session"])
+        self.assertNotEqual(entries[0]["subpath"], entries[1]["subpath"])
+
+    async def test_waits_for_readiness_after_concurrent_volume_creation(self):
+        self.client.volume.get.side_effect = [DaytonaConflictError("already created"),
+                                              SimpleNamespace(id="volume-id", state="pending_create"),
+                                              SimpleNamespace(id="volume-id", state="creating"), self.volume]
+        with patch("diffex_terminal_bench.daytona_backend.asyncio.sleep"):
+            await self.environment._create_sandbox(CreateSandboxFromImageParams(image=Image.base("ubuntu:24.04")), self.client)
+        self.create.assert_awaited_once()
+        self.assertEqual(self.client.volume.get.await_count, 4)
+
+    async def test_unusable_volume_prevents_sandbox_creation(self):
+        self.volume.state = "error"
+        with self.assertRaisesRegex(RuntimeError, "not usable"):
+            await self.environment._create_sandbox(CreateSandboxFromSnapshotParams(snapshot="fixture"), self.client)
+        self.create.assert_not_awaited()
+
+    async def test_volume_quota_error_is_not_retried_or_bypassed(self):
+        self.client.volume.get.side_effect = DaytonaError("volume quota", status_code=400)
+        with self.assertRaises(DaytonaError):
+            await self.environment._create_sandbox(CreateSandboxFromSnapshotParams(snapshot="fixture"), self.client)
+        self.assertEqual(self.client.volume.get.await_count, 1)
+        self.create.assert_not_awaited()
+
+    async def test_provisioning_wait_is_bounded(self):
+        self.volume.state = "creating"
+        timeout = asyncio.timeout
+        with patch("diffex_terminal_bench.daytona_backend.asyncio.timeout", side_effect=lambda seconds: timeout(0.01)):
+            with self.assertRaises(TimeoutError):
+                await self.environment._create_sandbox(CreateSandboxFromSnapshotParams(snapshot="fixture"), self.client)
+        self.create.assert_not_awaited()
+
+    async def test_dind_client_manager_and_custom_volume(self):
+        self.environment._client_manager = SimpleNamespace(get_client=AsyncMock(return_value=self.client))
+        self.environment._kwargs["artifact_volume"] = "dedicated-project-logs"
+        params = CreateSandboxFromImageParams(image=Image.base("docker:28.3.3-dind"))
+        await self.environment._create_sandbox(params)
+        self.client.volume.get.assert_awaited_once_with("dedicated-project-logs", create=True)
+        self.create.assert_awaited_once_with(params=params, daytona=self.client)
+
+    def test_native_full_job_selects_volume_backend(self):
+        config = JobConfig.model_validate(yaml.safe_load((Path(__file__).resolve().parents[1] / "full.yaml").read_text()))
+        self.assertEqual(config.environment.import_path, "diffex_terminal_bench.daytona_backend:RecordedDaytonaEnvironment")
+        self.assertEqual(config.environment.kwargs["artifact_volume"], "diffex-terminal-bench-logs")
+
+
+if __name__ == "__main__":
+    unittest.main()

From a3d26d97c7110cbcb4475ce0ce7d3aa84528e347 Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Thu, 24 Sep 2026 09:15:17 -0400
Subject: [PATCH 6/8] fix(coding-agent): await cleanup evidence

Allow the bounded evidence retries to span the remote cleanup grace period after a lost response. Preserve precise artifact failure phases.
---
 .../src/diffex_terminal_bench/adapter.py         |  2 +-
 .../src/diffex_terminal_bench/transport.py       |  4 +++-
 evals/terminal-bench/tests/test_recovery.py      | 16 +++++++++++-----
 3 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/evals/terminal-bench/src/diffex_terminal_bench/adapter.py b/evals/terminal-bench/src/diffex_terminal_bench/adapter.py
index 950ed2ec8..75adabca5 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/adapter.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/adapter.py
@@ -275,7 +275,7 @@ async def fetch_evidence():
                            or (outcome["status"] == "configuration_failure"
                                and not all((directory / name).is_file() for name in ("requested.json", "effective.json"))))
                 if cleanup is not None and not execution["interruption"] and unknown:
-                    outcome.update(status="artifact_collection_failure", reason="log_collection_failed")
+                    outcome.update(status="artifact_collection_failure", reason=f"{artifact_errors[0]['phase']}_failed")
         if failures:
             outcome["recovery_errors"] = failures
         context.metadata = {**(context.metadata or {}), "diffex_outcome": outcome, "diffex_run_dir": str(run_dir)}
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/transport.py b/evals/terminal-bench/src/diffex_terminal_bench/transport.py
index d7b81c8d9..062176429 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/transport.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/transport.py
@@ -21,7 +21,9 @@ async def retry_transport[T](operation: Callable[[], Awaitable[T]]) -> T:
                     transient |= exc.status_code in (408, 429, 500, 502, 503, 504)
                 if not transient or attempt == 2:
                     raise
-                delay = 2 ** attempt + random.uniform(0, 0.25)
+                # Cleanup may still be finishing its seven-second TERM/KILL
+                # cycle after a lost RPC response. Cover that publication lag.
+                delay = 4 * 2 ** attempt + random.uniform(0, 0.25)
                 if isinstance(exc, DaytonaError):
                     for key, value in exc.headers.items():
                         if key.lower().startswith("retry-after"):
diff --git a/evals/terminal-bench/tests/test_recovery.py b/evals/terminal-bench/tests/test_recovery.py
index e955aae0a..ee8baca2e 100644
--- a/evals/terminal-bench/tests/test_recovery.py
+++ b/evals/terminal-bench/tests/test_recovery.py
@@ -277,6 +277,11 @@ async def execute(**kwargs):
 
     async def test_pending_cleanup_publication_is_retried(self):
         attempts = 0
+        elapsed = 0
+
+        async def delay(seconds):
+            nonlocal elapsed
+            elapsed += seconds
 
         async def execute(**kwargs):
             nonlocal attempts
@@ -284,14 +289,14 @@ async def execute(**kwargs):
                 raise ConnectionError("response lost while cleanup continues")
             if "evidence.cjs" in kwargs["command"]:
                 attempts += 1
-                if attempts == 1:
+                if elapsed < 7:
                     return ExecResult(return_code=0, stdout="null", stderr="")
             return self.exec_result(kwargs)
 
         self.environment.exec.side_effect = execute
-        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+        with patch("diffex_terminal_bench.transport.asyncio.sleep", side_effect=delay):
             await self.agent.run("Task", self.environment, self.context)
-        self.assertEqual(attempts, 2)
+        self.assertEqual(attempts, 3)
         self.assertTrue(self.context.metadata["diffex_outcome"]["agent_execution_stopped"])
 
     async def test_local_artifact_write_failure_preserves_shutdown(self):
@@ -329,12 +334,13 @@ async def test_transport_retry_respects_rate_limit_and_permanent_errors(self):
         for error, count in ((DaytonaAuthenticationError("denied", status_code=401), 1),
                              (DaytonaError("limited", status_code=429, headers={"Retry-After-authenticated": "5"}), 3)):
             operation = create_autospec(self.agent._finish_run, side_effect=error)
-            with patch("diffex_terminal_bench.transport.asyncio.sleep") as sleep:
+            with patch("diffex_terminal_bench.transport.asyncio.sleep") as sleep, \
+                    patch("diffex_terminal_bench.transport.random.uniform", return_value=0):
                 with self.assertRaises(type(error)):
                     await retry_transport(lambda: operation(None, None, None, None))
             self.assertEqual(operation.await_count, count)
             if count == 3:
-                self.assertEqual([call.args[0] for call in sleep.call_args_list], [5, 5])
+                self.assertEqual([call.args[0] for call in sleep.call_args_list], [5, 8])
 
 
 if __name__ == "__main__":

From 606deae6abde024cca7b8e85e8e5a2d650c4e0b7 Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Thu, 24 Sep 2026 09:43:37 -0400
Subject: [PATCH 7/8] feat(coding-agent): prune completed trial logs

- Reclaim isolated volume prefixes after verification and local artifact validation.
- Retain recovery data when execution, collection, or sandbox teardown is incomplete.
---
 evals/terminal-bench/ARTIFACTS.md             |  17 +-
 evals/terminal-bench/README.md                |   1 +
 .../src/diffex_terminal_bench/benchmark.py    |   2 +
 .../src/diffex_terminal_bench/prune_volume.py |  42 +++
 .../src/diffex_terminal_bench/retention.py    | 172 ++++++++++++
 .../src/diffex_terminal_bench/transport.py    |   6 +-
 evals/terminal-bench/tests/test_retention.py  | 247 ++++++++++++++++++
 7 files changed, 483 insertions(+), 4 deletions(-)
 create mode 100644 evals/terminal-bench/src/diffex_terminal_bench/prune_volume.py
 create mode 100644 evals/terminal-bench/src/diffex_terminal_bench/retention.py
 create mode 100644 evals/terminal-bench/tests/test_retention.py

diff --git a/evals/terminal-bench/ARTIFACTS.md b/evals/terminal-bench/ARTIFACTS.md
index 5f1884673..cc75949f2 100644
--- a/evals/terminal-bench/ARTIFACTS.md
+++ b/evals/terminal-bench/ARTIFACTS.md
@@ -35,7 +35,22 @@ Keep job concurrency within the organization's remaining capacity, accounting fo
 See the [Daytona limits documentation](https://www.daytona.io/docs/en/limits/).
 This change does not increase job concurrency or sandbox resources.
 
-The shared volume is intentionally retained when a trial deletes its sandbox, including when local download fails.
+The staged runner automatically prunes a completed Diffex trial's volume prefixes after qualification, before releasing its concurrency slot.
+The native Harbor command in the README enables the same behavior with `--plugin diffex_terminal_bench.retention:VolumeCleanupPlugin`.
+Harbor 0.23.0 ignores plugins in YAML, so include this CLI option in other native run commands too.
+Cleanup requires a completed trial and verifier result, matching local reward and verifier logs, valid native shutdown and execution evidence, a valid trajectory, and no artifact errors or native session gaps.
+A zero reward is eligible; failed, interrupted, oracle-only, and incomplete trials retain their remote logs.
+All recorded trial sandboxes must be absent before deletion, including separate verifiers and retained task services.
+One short-lived Python sandbox mounts only that trial's exact prefixes, compares every remote file's SHA-256 against the local artifact copy, and empties the prefixes only if all files match.
+Unknown files, symlinks, or mismatched files preserve remote data for recovery.
+Local logs, transcripts, and results remain available.
+`volume-cleanup.json` records the decision, prefix locations, deleted file count, and cleanup sandbox deletion status.
+Cleanup failures do not change verifier results and retain recovery locations; transport operations use bounded retries.
+Remote comparison and pruning allow three attempts within 210 seconds, with a 60-second execution limit per attempt.
+The cleanup sandbox is deleted in a shielded finalizer, with a two-minute idle stop, immediate deletion after stopping, and a ten-minute maximum lifetime as fallbacks.
+This uses one extra sandbox creation per eligible trial, within the existing worker slot; include that overhead in the Tier 4 creation-rate and compute budget.
+The shared volume itself remains available to subsequent trials and concurrent jobs.
+
 To recover artifacts, select the relevant session in `artifact-volume.json`, mount its volume ID and exact subpath on a temporary sandbox, and download `/logs`.
 Delete that recovery sandbox afterward.
 Remove only a trial's recorded subpath when its artifacts are no longer needed; do not delete the shared volume while any job uses it.
diff --git a/evals/terminal-bench/README.md b/evals/terminal-bench/README.md
index 46277fd38..e5abe3e91 100644
--- a/evals/terminal-bench/README.md
+++ b/evals/terminal-bench/README.md
@@ -46,6 +46,7 @@ TB_STORAGE_MB=20480
 mkdir -p "$TB_OUTPUT"
 git rev-parse HEAD > "$TB_OUTPUT/adapter-commit.txt"
 TB_RUN=(uv run --locked harbor run --config full.yaml
+  --plugin diffex_terminal_bench.retention:VolumeCleanupPlugin
   --override-storage-mb "$TB_STORAGE_MB"
   --jobs-dir "$TB_OUTPUT" --job-name full)
 "${TB_RUN[@]}" --print-config > "$TB_OUTPUT/planned-config.json"
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/benchmark.py b/evals/terminal-bench/src/diffex_terminal_bench/benchmark.py
index 9e642bda8..5356f9f53 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/benchmark.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/benchmark.py
@@ -25,6 +25,7 @@
 from diffex_terminal_bench.daytona_backend import confirm_cleanup
 from diffex_terminal_bench.diagnostics import TrialExecutionFailure, describe_failure, diagnose_trial
 from diffex_terminal_bench.execution import PROVIDER_ENV
+from diffex_terminal_bench.retention import cleanup_trial_volume
 from diffex_terminal_bench.gate import STAGES, fingerprint, inventory, read_json, require_previous, sample_tasks, sha256, stage_tasks, validate_trial, verify_stage, write_json
 from diffex_terminal_bench.summary import refresh_index, write_index
 
@@ -174,6 +175,7 @@ async def run_trial(index: int, task: str) -> None:
                         if cancelled:
                             raise asyncio.CancelledError
                 evidence = validate_trial(trial_dir, task, identity, oracle=stage == "oracle")
+                await cleanup_trial_volume(client, trial_dir)
                 report["trials"].append({"task": task, "directory": trial_dir.name,
                                          "evidence": evidence, "files": inventory(trial_dir)})
                 report["trials"].sort(key=lambda row: row["directory"])
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/prune_volume.py b/evals/terminal-bench/src/diffex_terminal_bench/prune_volume.py
new file mode 100644
index 000000000..0fa029d3a
--- /dev/null
+++ b/evals/terminal-bench/src/diffex_terminal_bench/prune_volume.py
@@ -0,0 +1,42 @@
+"""Run inside a disposable sandbox with only the completed trial's prefixes."""
+
+import hashlib
+import json
+from pathlib import Path
+import stat
+import sys
+
+
+def prune(roots: list[Path], expected: dict[str, str]) -> int:
+    # Check every prefix before deleting anything. Never follow remote links.
+    files = []
+    directories = []
+    for root in roots:
+        if root.is_symlink() or not root.is_dir():
+            raise ValueError("Missing volume mount")
+        for path in root.rglob("*"):
+            mode = path.lstat().st_mode
+            if stat.S_ISDIR(mode):
+                directories.append(path)
+                continue
+            if not stat.S_ISREG(mode):
+                raise ValueError("Volume contains a link or special file")
+            relative = path.relative_to(root).as_posix()
+            with path.open("rb") as stream:
+                digest = hashlib.file_digest(stream, "sha256").hexdigest()
+            if expected.get(relative) != digest:
+                raise ValueError("Volume file has no matching local copy")
+            files.append(path)
+    for path in files:
+        path.unlink()
+    for path in sorted(directories, key=lambda item: len(item.parts), reverse=True):
+        path.rmdir()
+    if any(any(root.iterdir()) for root in roots):
+        raise ValueError("Volume prefix is not empty after cleanup")
+    return len(files)
+
+
+if __name__ == "__main__":
+    request = json.loads(Path(sys.argv[1]).read_text())
+    count = prune([Path(root) for root in request["roots"]], request["expected"])
+    print(json.dumps({"deleted_files": count, "confirmed": True}))
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/retention.py b/evals/terminal-bench/src/diffex_terminal_bench/retention.py
new file mode 100644
index 000000000..d4fe8be5d
--- /dev/null
+++ b/evals/terminal-bench/src/diffex_terminal_bench/retention.py
@@ -0,0 +1,172 @@
+"""Reclaim completed trial prefixes after verification and local validation."""
+
+import asyncio
+import json
+import logging
+import math
+from pathlib import Path
+import re
+from uuid import uuid4
+
+from daytona import AsyncDaytona, CreateSandboxFromSnapshotParams, VolumeMount
+from daytona.common.errors import DaytonaNotFoundError
+from harbor.job import Job
+from harbor.models.job.result import JobResult
+from harbor.trial.hooks import TrialEvent, TrialHookEvent
+from harbor.utils.trajectory_validator import TrajectoryValidator
+
+from diffex_terminal_bench.gate import inventory, read_json, write_json
+from diffex_terminal_bench.outcome import parse_outcome
+from diffex_terminal_bench.transport import retry_transport
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+def validate_retention(directory: Path) -> dict[str, str]:
+    result = read_json(directory / "result.json")
+    if result.get("exception_info") or not result.get("finished_at"):
+        raise ValueError("Trial did not complete")
+    rewards = (result.get("verifier_result") or {}).get("rewards")
+    if not isinstance(rewards, dict) or not rewards or any(
+        type(value) not in (int, float) or not math.isfinite(value) for value in rewards.values()
+    ):
+        raise ValueError("Verification did not complete")
+    verifier = directory / "verifier"
+    saved = read_json(verifier / "reward.json") if (verifier / "reward.json").exists() else {
+        "reward": float((verifier / "reward.txt").read_text())}
+    if saved != rewards or not (verifier / "test-stdout.txt").is_file():
+        raise ValueError("Verifier evidence is incomplete")
+    runs = list((directory / "agent").glob("run-*"))
+    if len(runs) != 1:
+        raise ValueError("Expected one completed Diffex run")
+    outcome = read_json(runs[0] / "outcome.json")
+    if outcome.get("status") != "completed" or outcome.get("artifact_errors") or outcome.get("recovery_errors"):
+        raise ValueError("Execution or artifact collection failed")
+    native = parse_outcome(runs[0], launcher_exit_code=outcome.get("launcher_exit_code"))
+    if native["status"] != "completed" or not native["cleanup_confirmed"]:
+        raise ValueError("Native completion and shutdown are not established")
+    if not TrajectoryValidator().validate(directory / "agent/trajectory.json"):
+        raise ValueError("Invalid local trajectory")
+    if read_json(runs[0] / "accounting.json").get("gaps"):
+        raise ValueError("Incomplete native sessions")
+    files = inventory(directory)
+    # Harbor stores the convention /logs/artifacts directory by source path.
+    expected = {}
+    for name, digest in files.items():
+        if name.startswith(("agent/", "verifier/", "user-agent/")):
+            expected[name] = digest
+        elif name.startswith("artifacts/logs/artifacts/"):
+            expected[name.removeprefix("artifacts/logs/")] = digest
+    return expected
+
+
+async def cleanup_trial_volume(client: AsyncDaytona, directory: Path) -> dict:
+    report = {"status": "retained", "confirmed": False}
+    sandbox = None
+    try:
+        expected = validate_retention(directory)
+        mounts = read_json(directory / "artifact-volume.json")["mounts"]
+        created = read_json(directory / "sandbox-created.json")["sandboxes"]
+        if not mounts or len(mounts) != len(created):
+            raise ValueError("Incomplete sandbox creation records")
+        for mount in mounts:
+            if not re.fullmatch(r"trials/[0-9a-f]{32}", mount["subpath"]) or mount["mount_path"] != "/logs":
+                raise ValueError("Invalid trial volume prefix")
+        # Retained services and separate verifiers must all be gone first.
+        for row in created:
+            async def confirm_absent():
+                try:
+                    await client.get(row["id"], request_timeout=10)
+                except DaytonaNotFoundError:
+                    return
+                # Harbor's delete request returns before deletion settles.
+                raise TimeoutError("Trial sandbox deletion is still pending")
+
+            await retry_transport(confirm_absent)
+        roots = [f"/mnt/trial-{index}" for index in range(len(mounts))]
+        report["mounts"] = mounts
+        report["cleanup_label"] = uuid4().hex
+        report["status"] = "cleaning"
+        write_json(directory / "volume-cleanup.json", report)
+        # Never retry a create: a lost response could allocate another sandbox.
+        # Automatic stop/delete also bounds leaks if the runner loses this RPC.
+        sandbox = await client.create(CreateSandboxFromSnapshotParams(
+            language="python", os_user="root", auto_stop_interval=2, auto_delete_interval=0, ttl_minutes=10,
+            labels={"diffex.volume_cleanup": report["cleanup_label"]},
+            volumes=[VolumeMount(volume_id=mount["volume_id"], subpath=mount["subpath"], mount_path=root)
+                     for mount, root in zip(mounts, roots, strict=True)],
+        ), timeout=60)
+        report["sandbox_id"] = sandbox.id
+        write_json(directory / "volume-cleanup.json", report)
+        await retry_transport(lambda: sandbox.fs.upload_file(
+            Path(__file__).with_name("prune_volume.py").read_bytes(), "/tmp/prune-volume.py"))
+        payload = json.dumps({"roots": roots, "expected": expected}).encode()
+        await retry_transport(lambda: sandbox.fs.upload_file(payload, "/tmp/prune-request.json"))
+
+        async def prune():
+            response = await sandbox.process.exec("python3 /tmp/prune-volume.py /tmp/prune-request.json", timeout=60)
+            if response.exit_code != 0:
+                raise ValueError("Remote artifact comparison or cleanup failed")
+            value = json.loads(response.result)
+            if value.get("confirmed") is not True:
+                raise ValueError("Remote cleanup was not confirmed")
+            return value
+
+        report.update(await retry_transport(prune, request_timeout=65, total_timeout=210))
+        report["status"] = "cleaned"
+    except Exception as exc:
+        report["status"] = "retained" if sandbox is None else "cleanup_failed"
+        report["error"] = type(exc).__name__
+        LOGGER.info("Retaining volume recovery records for %s: %s", directory.name, exc)
+    finally:
+        if sandbox is not None:
+            async def delete():
+                try:
+                    await client.delete(sandbox, wait=True, timeout=15)
+                except DaytonaNotFoundError:
+                    pass
+
+            deletion = asyncio.create_task(retry_transport(delete))
+            cancelled = False
+            while not deletion.done():
+                try:
+                    await asyncio.shield(deletion)
+                except asyncio.CancelledError:
+                    cancelled = True
+                except Exception:
+                    break
+            try:
+                deletion.result()
+                report["cleanup_sandbox_deleted"] = True
+            except Exception as exc:
+                report["cleanup_sandbox_deleted"] = False
+                report["sandbox_error"] = type(exc).__name__
+            if cancelled:
+                report["interrupted"] = True
+        try:
+            write_json(directory / "volume-cleanup.json", report)
+        except OSError:
+            LOGGER.warning("Could not save volume cleanup report for %s", directory, exc_info=True)
+        if sandbox is not None and cancelled:
+            raise asyncio.CancelledError
+    return report
+
+
+class VolumeCleanupPlugin:
+    async def on_job_start(self, job: Job) -> None:
+        job.add_hook(TrialEvent.END, self.on_trial_end)
+
+    async def on_trial_end(self, event: TrialHookEvent) -> None:
+        if (event.config.environment.import_path != "diffex_terminal_bench.daytona_backend:RecordedDaytonaEnvironment"
+                or not event.config.environment.delete):
+            return
+        directory = event.config.trials_dir / event.config.trial_name
+        try:
+            async with AsyncDaytona() as client:
+                await cleanup_trial_volume(client, directory)
+        except Exception:
+            LOGGER.warning("Volume cleanup unavailable for %s", directory, exc_info=True)
+
+    async def on_job_end(self, job_result: JobResult) -> None:
+        pass
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/transport.py b/evals/terminal-bench/src/diffex_terminal_bench/transport.py
index 062176429..358f1da19 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/transport.py
+++ b/evals/terminal-bench/src/diffex_terminal_bench/transport.py
@@ -7,13 +7,13 @@
 from daytona.common.errors import DaytonaConnectionError, DaytonaError, DaytonaTimeoutError
 
 
-async def retry_transport[T](operation: Callable[[], Awaitable[T]]) -> T:
+async def retry_transport[T](operation: Callable[[], Awaitable[T]], *, request_timeout: float = 20, total_timeout: float = 90) -> T:
     # The outer bound includes backoff and all requests; each request is also
     # bounded so a lost connection cannot consume the entire retry budget.
-    async with asyncio.timeout(90):
+    async with asyncio.timeout(total_timeout):
         for attempt in range(3):
             try:
-                async with asyncio.timeout(20):
+                async with asyncio.timeout(request_timeout):
                     return await operation()
             except Exception as exc:
                 transient = isinstance(exc, (TimeoutError, ConnectionError, DaytonaConnectionError, DaytonaTimeoutError))
diff --git a/evals/terminal-bench/tests/test_retention.py b/evals/terminal-bench/tests/test_retention.py
new file mode 100644
index 000000000..69ccca91a
--- /dev/null
+++ b/evals/terminal-bench/tests/test_retention.py
@@ -0,0 +1,247 @@
+"""Completed Harbor trials reclaim only fully retained volume artifacts."""
+
+import asyncio
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+from types import SimpleNamespace
+import unittest
+from unittest.mock import AsyncMock, Mock, create_autospec, patch
+
+from daytona.common.errors import DaytonaNotFoundError
+from harbor.cli.job_plugins import attach_job_plugin
+from harbor.trial.hooks import TrialEvent
+from harbor.trial.single_step import SingleStepTrial
+from harbor.trial.trial import Trial
+
+from diffex_terminal_bench.gate import read_json, write_json
+from diffex_terminal_bench.retention import cleanup_trial_volume, validate_retention
+import test_gate
+
+
+class RetentionTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        fixture = test_gate.GateTests()
+        fixture.setUp()
+        self.addCleanup(fixture.doCleanups)
+        self.directory = fixture.directory
+        self.run = fixture.run
+        self.remote = fixture.root / "volume"
+        self.remote.mkdir()
+        self.mounts = [{"volume_id": "shared", "mount_path": "/logs", "subpath": "trials/" + "a" * 32}]
+        write_json(self.directory / "artifact-volume.json", {"mounts": self.mounts})
+        for name in ("agent", "verifier"):
+            shutil.copytree(self.directory / name, self.remote / name)
+        self.uploads = {}
+
+        async def upload(data, path):
+            self.uploads[path] = data
+
+        async def execute(command, timeout):
+            # Execute the actual uploaded cleanup program against isolated
+            # temporary prefix directories, replacing only the mount boundary.
+            request = json.loads(self.uploads["/tmp/prune-request.json"])
+            request["roots"] = [str(self.remote)] + ([str(self.verifier_remote)] if len(self.mounts) == 2 else [])
+            script = fixture.root / "prune.py"
+            script.write_bytes(self.uploads["/tmp/prune-volume.py"])
+            path = fixture.root / "request.json"
+            write_json(path, request)
+            result = subprocess.run([sys.executable, str(script), str(path)], capture_output=True, text=True, timeout=timeout)
+            return SimpleNamespace(exit_code=result.returncode, result=result.stdout)
+
+        self.sandbox = SimpleNamespace(id="cleanup", fs=SimpleNamespace(upload_file=AsyncMock(side_effect=upload)),
+                                       process=SimpleNamespace(exec=AsyncMock(side_effect=execute)))
+        self.client = SimpleNamespace(get=AsyncMock(side_effect=DaytonaNotFoundError("gone")),
+                                      create=AsyncMock(return_value=self.sandbox), delete=AsyncMock())
+
+    async def test_completed_zero_reward_reclaims_remote_only(self):
+        sibling = self.remote.parent / "other-trial"
+        sibling.mkdir()
+        (sibling / "live.log").write_text("still running")
+        report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertEqual(report["status"], "cleaned")
+        self.assertTrue(report["confirmed"])
+        self.assertGreater(report["deleted_files"], 0)
+        self.assertEqual(list(self.remote.iterdir()), [])
+        self.assertTrue((self.run / "sessions/root.jsonl").exists())
+        self.assertEqual((sibling / "live.log").read_text(), "still running")
+        request = self.client.create.call_args.args[0]
+        self.assertEqual(request.volumes[0].subpath, self.mounts[0]["subpath"])
+        self.assertEqual(request.auto_delete_interval, 0)
+        self.assertEqual(request.ttl_minutes, 10)
+        self.client.delete.assert_awaited_once_with(self.sandbox, wait=True, timeout=15)
+
+    async def test_retains_failed_unverified_and_incomplete_trials(self):
+        changes = [(self.directory / "result.json", "finished_at", None),
+                   (self.directory / "result.json", "exception_info", {"exception_type": "VerifierTimeoutError"}),
+                   (self.directory / "result.json", "verifier_result", None),
+                   (self.run / "outcome.json", "artifact_errors", [{"phase": "download"}]),
+                   (self.run / "cleanup.json", "agent_execution_stopped", False),
+                   (self.run / "accounting.json", "gaps", ["missing_session"])]
+        for path, key, value in changes:
+            with self.subTest(key=key):
+                original = path.read_bytes()
+                data = read_json(path)
+                data[key] = value
+                write_json(path, data)
+                report = await cleanup_trial_volume(self.client, self.directory)
+                self.assertEqual(report["status"], "retained")
+                self.client.create.assert_not_awaited()
+                path.write_bytes(original)
+
+    async def test_hash_mismatch_and_missing_local_files_prevent_any_deletion(self):
+        path = self.remote / "verifier/test-stdout.txt"
+        for value in ("not collected", ""):
+            path.write_text(value)
+            report = await cleanup_trial_volume(self.client, self.directory)
+            self.assertEqual(report["status"], "cleanup_failed")
+            self.assertTrue((self.remote / "agent/run-fixture/stdout.jsonl").exists())
+            self.assertTrue(report["cleanup_sandbox_deleted"])
+        path.write_bytes((self.directory / "verifier/test-stdout.txt").read_bytes())
+        (self.remote / "uncollected.log").write_text("remote only")
+        report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertFalse(report["confirmed"])
+        self.assertTrue((self.remote / "uncollected.log").exists())
+
+    async def test_links_never_followed_or_deleted(self):
+        (self.remote / "linked-home").symlink_to(self.directory, target_is_directory=True)
+        report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertEqual(report["status"], "cleanup_failed")
+        self.assertTrue((self.directory / "result.json").exists())
+        self.assertTrue((self.remote / "linked-home").is_symlink())
+
+    async def test_separate_verifier_prefixes_validated_together(self):
+        self.verifier_remote = self.remote.parent / "verifier-volume"
+        self.verifier_remote.mkdir()
+        shutil.move(self.remote / "verifier", self.verifier_remote / "verifier")
+        self.mounts.append({**self.mounts[0], "subpath": "trials/" + "b" * 32})
+        write_json(self.directory / "artifact-volume.json", {"mounts": self.mounts})
+        write_json(self.directory / "sandbox-created.json", {"sandboxes": [{"id": "agent"}, {"id": "verifier"}]})
+        report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertEqual(report["status"], "cleaned")
+        self.assertEqual(list(self.remote.iterdir()), [])
+        self.assertEqual(list(self.verifier_remote.iterdir()), [])
+        self.assertEqual(self.client.get.await_count, 2)
+        self.assertEqual(len(self.client.create.call_args.args[0].volumes), 2)
+
+    async def test_existing_sandbox_or_invalid_prefix_prevents_cleanup(self):
+        self.client.get.side_effect = None
+        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+            report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertEqual(report["status"], "retained")
+        self.client.create.assert_not_awaited()
+        self.client.get.side_effect = DaytonaNotFoundError("gone")
+        for prefix in ("trials/../other", "trials/", "", "/"):
+            self.mounts[0]["subpath"] = prefix
+            write_json(self.directory / "artifact-volume.json", {"mounts": self.mounts})
+            report = await cleanup_trial_volume(self.client, self.directory)
+            self.assertEqual(report["status"], "retained")
+            self.client.create.assert_not_awaited()
+
+    async def test_delayed_sandbox_deletion_is_polled_before_mounting(self):
+        self.client.get.side_effect = [SimpleNamespace(state="deleting"), DaytonaNotFoundError("gone")]
+        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+            report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertEqual(report["status"], "cleaned")
+        self.assertEqual(self.client.get.await_count, 2)
+
+    async def test_transient_upload_and_deletion_failures_retry(self):
+        upload = self.sandbox.fs.upload_file.side_effect
+        attempts = 0
+
+        async def flaky(data, path):
+            nonlocal attempts
+            attempts += 1
+            if attempts == 1:
+                raise ConnectionError("lost transport")
+            await upload(data, path)
+
+        self.sandbox.fs.upload_file.side_effect = flaky
+        self.client.delete.side_effect = [ConnectionError("lost response"), None]
+        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+            report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertEqual(report["status"], "cleaned")
+        self.assertEqual(self.client.delete.await_count, 2)
+        self.assertEqual(attempts, 3)
+
+    async def test_lost_cleanup_response_retries_idempotently(self):
+        execute = self.sandbox.process.exec.side_effect
+        attempts = 0
+
+        async def lose_response(command, timeout):
+            nonlocal attempts
+            attempts += 1
+            response = await execute(command, timeout)
+            if attempts == 1:
+                raise ConnectionError("response lost after deletion")
+            return response
+
+        self.sandbox.process.exec.side_effect = lose_response
+        with patch("diffex_terminal_bench.transport.asyncio.sleep"):
+            report = await cleanup_trial_volume(self.client, self.directory)
+        self.assertTrue(report["confirmed"])
+        self.assertEqual(attempts, 2)
+        self.assertEqual(list(self.remote.iterdir()), [])
+
+    async def test_cancellation_waits_for_cleanup_sandbox_deletion(self):
+        deleting = asyncio.Event()
+        release = asyncio.Event()
+
+        async def delete(*args, **kwargs):
+            deleting.set()
+            await release.wait()
+
+        self.client.delete.side_effect = delete
+        self.sandbox.fs.upload_file.side_effect = asyncio.CancelledError
+        task = asyncio.create_task(cleanup_trial_volume(self.client, self.directory))
+        await asyncio.wait_for(deleting.wait(), 1)
+        task.cancel()
+        await asyncio.sleep(0)
+        self.assertFalse(task.done())
+        release.set()
+        with self.assertRaises(asyncio.CancelledError):
+            await task
+        self.assertTrue(read_json(self.directory / "volume-cleanup.json")["cleanup_sandbox_deleted"])
+        self.assertTrue((self.remote / "agent/run-fixture/stdout.jsonl").exists())
+
+    async def test_plugin_attaches_to_real_harbor_finalize_after_result_is_saved(self):
+        job = Mock()
+        plugin = await attach_job_plugin(job, "diffex_terminal_bench.retention:VolumeCleanupPlugin")
+        job.add_hook.assert_called_once_with(TrialEvent.END, plugin.on_trial_end)
+        result = read_json(self.directory / "result.json")
+        (self.directory / "result.json").unlink()
+        trial = create_autospec(SingleStepTrial, instance=True)
+        trial.paths = SimpleNamespace(result_path=self.directory / "result.json")
+        trial.result = Mock(model_dump_json=Mock(return_value=json.dumps(result)))
+        stopped = False
+
+        async def stop():
+            nonlocal stopped
+            stopped = True
+
+        async def emit(event):
+            self.assertTrue(stopped)
+            self.assertEqual(event, TrialEvent.END)
+            self.assertTrue(trial.paths.result_path.exists())
+            await plugin.on_trial_end(SimpleNamespace(config=SimpleNamespace(
+                environment=SimpleNamespace(delete=True, import_path="diffex_terminal_bench.daytona_backend:RecordedDaytonaEnvironment"),
+                trials_dir=self.directory.parent, trial_name=self.directory.name)))
+
+        trial._stop_agent_environment.side_effect = stop
+        trial._emit.side_effect = emit
+        with patch("diffex_terminal_bench.retention.AsyncDaytona") as factory:
+            factory.return_value.__aenter__.return_value = self.client
+            await Trial._finalize(trial)
+        self.assertEqual(list(self.remote.iterdir()), [])
+
+    def test_artifact_convention_paths_map_to_harbor_local_storage(self):
+        path = self.directory / "artifacts/logs/artifacts/file.txt"
+        path.parent.mkdir(parents=True)
+        path.write_text("task artifact")
+        self.assertIn("artifacts/file.txt", validate_retention(self.directory))
+
+
+if __name__ == "__main__":
+    unittest.main()

From 303f53a05181058b8d38ca6d9265a7ac2c598cdc Mon Sep 17 00:00:00 2001
From: Jay Yu <yibai14651@gmail.com>
Date: Thu, 24 Sep 2026 09:56:42 -0400
Subject: [PATCH 8/8] fix(coding-agent): avoid XZ in eval setup

- Use checksum-pinned gzip Node archives to avoid XZ package 404s during benchmark setup.
- Cover missing XZ and corrupt archives; verify installation on all 89 task images.
---
 .../src/diffex_terminal_bench/install.sh      | 14 +++++--
 .../src/diffex_terminal_bench/pins.toml       |  5 ++-
 .../terminal-bench/tests/test_installation.py | 42 +++++++++++++++----
 3 files changed, 49 insertions(+), 12 deletions(-)

diff --git a/evals/terminal-bench/src/diffex_terminal_bench/install.sh b/evals/terminal-bench/src/diffex_terminal_bench/install.sh
index f780199bc..b428648a1 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/install.sh
+++ b/evals/terminal-bench/src/diffex_terminal_bench/install.sh
@@ -21,6 +21,14 @@ case "$(uname -m)" in
 esac
 export NODE_ARCH="$arch" NODE_SHA256="$checksum"
 
+# Harbor's sandbox log transfers also require these base-image utilities.
+for utility in tar gzip; do
+    if ! command -v "$utility" >/dev/null; then
+        echo "Setup requires $utility in the task image" >&2
+        exit 1
+    fi
+done
+
 # Refuse to reuse an earlier install or overwrite a task-provided command.
 mkdir "$prefix"
 if [ -e /usr/local/bin/diffex ] || [ -L /usr/local/bin/diffex ]; then
@@ -30,7 +38,7 @@ fi
 mkdir -p "$HOME" "$prefix/node" "$prefix/cli" "$prefix/work"
 cd "$prefix/work"
 packages=()
-for package in ca-certificates wget xz-utils; do
+for package in ca-certificates wget; do
     if [ "$(dpkg-query -W -f='${Status}' "$package" 2>/dev/null || true)" != "install ok installed" ]; then
         packages+=("$package")
     fi
@@ -55,11 +63,11 @@ if [ "${#packages[@]}" -gt 0 ]; then
         sleep "$((attempt * 2))"
     done
 fi
-archive="node-v$NODE_VERSION-linux-$arch.tar.xz"
+archive="node-v$NODE_VERSION-linux-$arch.tar.gz"
 wget --https-only --tries=3 --timeout=20 --output-document="$archive" \
     "https://nodejs.org/dist/v$NODE_VERSION/$archive"
 printf '%s  %s\n' "$checksum" "$archive" | sha256sum --check --strict
-tar -xJf "$archive" -C "$prefix/node" --strip-components=1
+tar -xzf "$archive" -C "$prefix/node" --strip-components=1
 rm "$archive"
 export PATH="$prefix/node/bin:$PATH"
 test "$(node --version)" = "v$NODE_VERSION"
diff --git a/evals/terminal-bench/src/diffex_terminal_bench/pins.toml b/evals/terminal-bench/src/diffex_terminal_bench/pins.toml
index e104dd002..e9bebb083 100644
--- a/evals/terminal-bench/src/diffex_terminal_bench/pins.toml
+++ b/evals/terminal-bench/src/diffex_terminal_bench/pins.toml
@@ -8,5 +8,6 @@ release_agents_sha256 = "6f194cd7f04d80e08cd3c8b22096de33ba5dbb2ba42176567c46657
 setup_timeout_sec = 600
 
 [node_sha256]
-x64 = "d60acfe00a2932254bb0ad20e01b0d74397a0875595de719654b214f4b03f307"
-arm64 = "fff4078c5def658577f92c88db7db3bc0072924bfb93fe52c1e744a54e94abb8"
+# Official node-v22.23.2-linux-<arch>.tar.gz archives.
+x64 = "b294a556e639d64338823920e5866c21c02741742d2e1529ee1a225c1ec9252a"
+arm64 = "013b59cfd2819703a6f4a14ab891fc46fc2a4e3f5bcd92de3fb4929b43e35b30"
diff --git a/evals/terminal-bench/tests/test_installation.py b/evals/terminal-bench/tests/test_installation.py
index bc965e9d4..c5baa465c 100644
--- a/evals/terminal-bench/tests/test_installation.py
+++ b/evals/terminal-bench/tests/test_installation.py
@@ -4,6 +4,7 @@
 import json
 import os
 from pathlib import Path
+import shutil
 import subprocess
 import sys
 import tempfile
@@ -111,7 +112,8 @@ def test_registry_metadata_must_match_all_pins(self):
 
 
 class PackageInstallTests(unittest.TestCase):
-    def run_install(self, *, installed=False, update_failures=0, install_failures=0):
+    def run_install(self, *, installed=False, missing_package="wget", update_failures=0,
+                    install_failures=0, corrupt_archive=False):
         with tempfile.TemporaryDirectory() as temp:
             root = Path(temp)
             commands = root / "bin"
@@ -123,6 +125,7 @@ def run_install(self, *, installed=False, update_failures=0, install_failures=0)
 import json
 import os
 from pathlib import Path
+import subprocess
 import sys
 
 root = Path(os.environ["INSTALL_TEST_ROOT"])
@@ -133,7 +136,7 @@ def run_install(self, *, installed=False, update_failures=0, install_failures=0)
 if name == "uname":
     print("Linux" if args == ["-s"] else "x86_64")
 elif name == "dpkg-query":
-    if os.environ["INSTALLED"] == "1" or args[-1] != "xz-utils":
+    if os.environ["INSTALLED"] == "1" or args[-1] != os.environ["MISSING_PACKAGE"]:
         print("install ok installed", end="")
     else:
         sys.exit(1)
@@ -146,11 +149,20 @@ def run_install(self, *, installed=False, update_failures=0, install_failures=0)
         print("404 Not Found" if operation == "install" else "Failed to fetch index", file=sys.stderr)
         sys.exit(100)
 elif name == "wget":
-    # Stop after dependency setup; never download or install real packages.
-    sys.exit(42)
+    # Never download or install real packages.
+    if os.environ["CORRUPT_ARCHIVE"] == "1":
+        filename = next(arg.split("=", 1)[1] for arg in args if arg.startswith("--output-document="))
+        Path(filename).write_bytes(b"corrupt Node archive")
+    else:
+        sys.exit(42)
+elif name == "tar":
+    sys.exit(43)
+elif name == "sha256sum":
+    # macOS sha256sum lacks GNU's --strict; retain real checksum verification.
+    sys.exit(subprocess.run([os.environ["SHA256SUM"], "-c", "-"], input=sys.stdin.read(), text=True).returncode)
 ''')
             stub.chmod(0o755)
-            for name in ("uname", "dpkg-query", "apt-get", "wget", "sleep"):
+            for name in ("uname", "dpkg-query", "apt-get", "wget", "sleep", "tar", "sha256sum"):
                 (commands / name).symlink_to(stub)
             with as_file(files("diffex_terminal_bench").joinpath("install.sh")) as source:
                 # Redirect installation paths and capture logs without /dev/fd process substitution.
@@ -166,6 +178,8 @@ def run_install(self, *, installed=False, update_failures=0, install_failures=0)
                 "NODE_SHA256_X64": PINS["node_sha256"]["x64"], "INSTALL_TEST_ROOT": temp,
                 "INSTALLED": str(int(installed)), "UPDATE_FAILURES": str(update_failures),
                 "INSTALL_FAILURES": str(install_failures),
+                "MISSING_PACKAGE": missing_package, "CORRUPT_ARCHIVE": str(int(corrupt_archive)),
+                "SHA256SUM": shutil.which("sha256sum"),
             })
             self.assertTrue((root / "calls.jsonl").exists(), result.stderr)
             calls = [json.loads(line) for line in (root / "calls.jsonl").read_text().splitlines()]
@@ -176,6 +190,20 @@ def test_installed_dependencies_do_not_need_package_network(self):
         self.assertEqual(result.returncode, 42, result.stderr)
         self.assertFalse(any(call[0] == "apt-get" for call in calls))
 
+    def test_missing_xz_does_not_need_apt_and_downloads_gzip(self):
+        result, calls = self.run_install(missing_package="xz-utils", install_failures=99)
+        self.assertEqual(result.returncode, 42, result.stderr)
+        self.assertFalse(any(call[0] == "apt-get" for call in calls))
+        download = next(call for call in calls if call[0] == "wget")
+        self.assertEqual(download[-1],
+                         f"https://nodejs.org/dist/v{PINS['node_version']}/node-v{PINS['node_version']}-linux-x64.tar.gz")
+
+    def test_corrupt_node_archive_is_rejected_before_extraction(self):
+        result, calls = self.run_install(installed=True, corrupt_archive=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("FAILED", result.stdout + result.stderr)
+        self.assertFalse(any(call[0] == "tar" for call in calls))
+
     def test_missing_archive_refreshes_indexes_and_retries(self):
         result, calls = self.run_install(install_failures=1)
         self.assertEqual(result.returncode, 42, result.stderr)
@@ -184,9 +212,9 @@ def test_missing_archive_refreshes_indexes_and_retries(self):
                          ["update", "install", "update", "install"])
         for call in apt:
             if "install" in call:
-                self.assertIn("xz-utils", call)
+                self.assertIn("wget", call)
                 self.assertNotIn("ca-certificates", call)
-                self.assertNotIn("wget", call)
+                self.assertNotIn("xz-utils", call)
 
     def test_index_failure_does_not_install_from_stale_indexes(self):
         result, calls = self.run_install(update_failures=1)