fix: a dead rollout no longer fails its task; serialize worktree setup

harness2/suite.py

@@ -523,13 +523,20 @@ def execute(
         runmark.mark_complete(run_dir, run_id=run_id, harness_dir=harness_dir)
         return run_dir
 
+    def _meta(self, run_dir: Path) -> dict:
+        """The rollout's meta.json, or {} for a rollout that died before writing one."""
+        try:
+            return json.loads((run_dir / "meta.json").read_text())
+        except (OSError, ValueError):
+            return {}
+
     def rollout_cost(self, run_dir: Path) -> float:
-        return float(json.loads((run_dir / "meta.json").read_text())["cost_usd"])
+        return float(self._meta(run_dir).get("cost_usd", 0.0))
 
     def delivered(self, run_dir: Path) -> bool:
-        if not (run_dir / "meta.json").is_file():
+        kind = self._meta(run_dir).get("deliverable")
+        if kind is None:
             return False
-        kind = json.loads((run_dir / "meta.json").read_text())["deliverable"]
         if kind == "files":
             return any(path.is_file() for path in (run_dir / "output").rglob("*"))
         path = run_dir / ("deliverable.patch" if kind == "diff" else "final.md")
@@ -541,7 +548,9 @@ def trajectory(self, run_dir: Path, *, obs_trunc: int) -> str:
         )
 
     def deliverable_files(self, run_dir: Path) -> dict[str, str]:
-        kind = json.loads((run_dir / "meta.json").read_text())["deliverable"]
+        kind = self._meta(run_dir).get("deliverable")
+        if kind is None:
+            return {}
         if kind == "files":
             paths = {
                 p.relative_to(run_dir / "output").as_posix(): p
@@ -565,8 +574,9 @@ def deliverable_files(self, run_dir: Path) -> dict[str, str]:
     def deliverable_text(self, run_dir: Path) -> str:
         files = self.deliverable_files(run_dir)
         text = "\n\n".join(files.values())
-        if "deliverable.patch" in files:
-            text += "\n\n" + (run_dir / "final.md").read_text(encoding="utf-8")
+        final = run_dir / "final.md"
+        if "deliverable.patch" in files and final.is_file():
+            text += "\n\n" + final.read_text(encoding="utf-8")
         return text
 
     def validate_tree(self, tree: dict[str, str]) -> None:

harness2/workspace.py

@@ -19,42 +19,49 @@
 import shutil
 import subprocess
 import tempfile
+import threading
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
 from pathlib import Path
 
 from .settings import ConfigError
 
+# Rollouts run in threads. `git worktree prune` racing another thread's `worktree add` can
+# delete a worktree that is still being created, so worktree bookkeeping is serialized.
+_WORKTREES = threading.Lock()
+
 
 @contextmanager
 def project_workspace(project: Path, ref: str) -> Iterator[Path]:
     ws = Path(tempfile.mkdtemp(prefix="harness2-ws-")).resolve()
     try:
-        pruned = subprocess.run(
-            ["git", "-C", str(project), "worktree", "prune"],
-            capture_output=True,
-            text=True,
-        )
-        if pruned.returncode:
-            raise ConfigError(f"cannot prune project worktrees: {pruned.stderr.strip()}")
-        result = subprocess.run(
-            ["git", "-C", str(project), "worktree", "add", "--detach", "--", str(ws), ref],
-            capture_output=True,
-            text=True,
-        )
-        if result.returncode:
-            raise ConfigError(
-                f"cannot create project workspace at {ref!r}: {result.stderr.strip()}"
-            )
+        with _WORKTREES:
+            _add_worktree(project, ws, ref)
         yield ws
     finally:
-        subprocess.run(
-            ["git", "-C", str(project), "worktree", "remove", "--force", str(ws)],
-            capture_output=True,
-        )
+        with _WORKTREES:
+            subprocess.run(
+                ["git", "-C", str(project), "worktree", "remove", "--force", str(ws)],
+                capture_output=True,
+            )
         shutil.rmtree(ws, ignore_errors=True)
 
 
+def _add_worktree(project: Path, ws: Path, ref: str) -> None:
+    pruned = subprocess.run(
+        ["git", "-C", str(project), "worktree", "prune"], capture_output=True, text=True
+    )
+    if pruned.returncode:
+        raise ConfigError(f"cannot prune project worktrees: {pruned.stderr.strip()}")
+    result = subprocess.run(
+        ["git", "-C", str(project), "worktree", "add", "--detach", "--", str(ws), ref],
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode:
+        raise ConfigError(f"cannot create project workspace at {ref!r}: {result.stderr.strip()}")
+
+
 @contextmanager
 def files_workspace(src: Path | None) -> Iterator[Path]:
     ws = Path(tempfile.mkdtemp(prefix="harness2-ws-")).resolve()

tests/test_suite.py

@@ -199,6 +199,13 @@ def test_module_task_protocol_splits_and_prompt_profiles(self):
         with self.assertRaises(ConfigError):
             self.adapter.tasks("missing")
 
+    def test_a_rollout_that_died_before_writing_anything_reads_as_empty(self):
+        dead = self.root / "missing"
+        self.assertFalse(self.adapter.delivered(dead))
+        self.assertEqual(self.adapter.deliverable_files(dead), {})
+        self.assertEqual(self.adapter.deliverable_text(dead), "")
+        self.assertEqual(self.adapter.rollout_cost(dead), 0.0)
+
     def test_claude_execute_capture_idempotence_and_isolation(self):
         self.assertFalse(self.adapter.delivered(self.root / "missing"))
         run_dir = self.adapter.execute("one", None, "nested/run")

tests/test_workspace.py

@@ -19,6 +19,7 @@
 import subprocess
 import tempfile
 import unittest
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 from harness2 import workspace
@@ -37,6 +38,18 @@ def setUp(self):
         (self.project / "CLAUDE.md").write_text("instructions\n")
         workspace.commit_all(self.project, "initial")
 
+    def test_concurrent_project_workspaces(self):
+        def open_one(_):
+            with workspace.project_workspace(self.project, "HEAD") as ws:
+                return (ws / "code.txt").read_text()
+
+        with ThreadPoolExecutor(8) as pool:
+            self.assertEqual(list(pool.map(open_one, range(16))), ["before\n"] * 16)
+        listed = subprocess.run(
+            ["git", "-C", str(self.project), "worktree", "list"], capture_output=True, text=True
+        )
+        self.assertEqual(len(listed.stdout.splitlines()), 1)
+
     def test_project_diff_apply_and_cleanup(self):
         with workspace.project_workspace(self.project, "HEAD") as ws:
             self.assertFalse(ws.is_relative_to(self.project))