fix(runner): Preserve isolated verification artifacts

safe_mini/observation/policies.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import json
-import os
 import subprocess
 from pathlib import Path
 
@@ -50,7 +49,7 @@ def final_tests_pass(cwd: str | Path, *, command: str = "python3 tests/run_tests
         shell=True,
         cwd=cwd_path,
         env={
-            "PATH": os.environ.get("PATH") or SANITIZED_PATH,
+            "PATH": SANITIZED_PATH,
             "PYTHONPATH": str(cwd_path),
             "HOME": str(cwd_path / ".agent-home"),
             "LANG": "C.UTF-8",

safe_mini/worktree.py

@@ -16,22 +16,26 @@ class WorktreeProvisioner:
     def __init__(self, base_repo: str | Path, *, tmp_root: str | Path | None = None) -> None:
         self.base_repo = Path(base_repo).resolve()
         self.tmp_root = Path(tmp_root).resolve() if tmp_root else None
-        self._tempdir: tempfile.TemporaryDirectory[str] | None = None
+        self._tempdir: Path | None = None
         self.path: Path | None = None
 
     def provision(self) -> Path:
         if not self.base_repo.exists() or not self.base_repo.is_dir():
             raise FileNotFoundError(f"base repo does not exist: {self.base_repo}")
 
         self.cleanup()
-        self._tempdir = tempfile.TemporaryDirectory(prefix="safe-mini-", dir=self.tmp_root)
-        target = Path(self._tempdir.name) / "repo"
-        ignore = shutil.ignore_patterns(".git", "__pycache__", ".pytest_cache", ".mypy_cache")
-        shutil.copytree(self.base_repo, target, ignore=ignore)
-        home = target / ".agent-home"
-        home.mkdir(mode=0o700, exist_ok=True)
-        self.path = target
-        return target
+        self._tempdir = Path(tempfile.mkdtemp(prefix="safe-mini-", dir=self.tmp_root))
+        try:
+            target = self._tempdir / "repo"
+            ignore = shutil.ignore_patterns(".git", "__pycache__", ".pytest_cache", ".mypy_cache")
+            shutil.copytree(self.base_repo, target, ignore=ignore)
+            home = target / ".agent-home"
+            home.mkdir(mode=0o700, exist_ok=True)
+            self.path = target
+            return target
+        except Exception:
+            self.cleanup()
+            raise
 
     def env_for(self, cwd: Path) -> dict[str, str]:
         return {
@@ -43,7 +47,8 @@ def env_for(self, cwd: Path) -> dict[str, str]:
 
     def cleanup(self) -> None:
         if self._tempdir is not None:
-            self._tempdir.cleanup()
+            if self._tempdir.exists():
+                shutil.rmtree(self._tempdir)
             self._tempdir = None
         self.path = None
 

tests/test_runner_integration.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 
+import safe_mini.worktree as worktree_module
 from safe_mini import Budget, Chunk, ExecutorPolicy, FailureClass, ObservationPolicy, SafeMiniRunner
 from tests._fixtures.scripted_model import ScriptedModel
 
@@ -25,13 +26,64 @@ def run_model(mode: str, *, budget: Budget | None = None, policy=ExecutorPolicy.
     )
 
 
-def test_success_path_completes():
+def test_success_path_completes_without_host_path(monkeypatch):
+    monkeypatch.setenv("PATH", "/host-path-that-must-not-be-used")
     result = run_model("success")
     assert result.success is True
     assert result.final_tests_pass is True
     assert result.failure_class is None
 
 
+def test_greeting_task_replays_in_fresh_worktrees_without_changing_source(
+    tmp_path: Path, monkeypatch
+):
+    """Exercise the concrete runner on the smallest JustAI-shaped coding task."""
+
+    source = tmp_path / "greeting-source"
+    tests = source / "tests"
+    tests.mkdir(parents=True)
+    (tests / "run_tests.py").write_text(
+        "from greeting import greet\n\nassert greet('Justin') == 'Hello, Justin!'\n"
+    )
+
+    monkeypatch.setattr(worktree_module.tempfile, "tempdir", str(tmp_path))
+    action = ScriptedModel._cmd(
+        "python3 - <<'PY'\n"
+        "from pathlib import Path\n"
+        "Path('greeting.py').write_text(\"def greet(name):\\n    return 'Hello, ' + name + '!'\\n\")\n"
+        "PY"
+    )
+    task = Chunk(
+        id="greeting",
+        description="Create greeting.py with greet(name).",
+        success_criteria="greet(name) returns Hello, <name>!",
+        budget=Budget(move_budget=2, observation_budget=4_000),
+    )
+
+    results = [
+        SafeMiniRunner(
+            ScriptedModel(responses=[action]),
+            repo_path=source,
+            keep_worktree=True,
+            final_check_command="python3 tests/run_tests.py",
+        ).run(task, task.budget, ObservationPolicy.FULL, ExecutorPolicy.SAFE)
+        for _ in range(2)
+    ]
+
+    assert [result.success for result in results] == [True, True]
+    assert [result.final_tests_pass for result in results] == [True, True]
+    assert [result.steps for result in results] == [1, 1]
+    assert not (source / "greeting.py").exists()
+
+    worktrees = [Path(result.worktree_path or "") for result in results]
+    assert len({path.resolve() for path in worktrees}) == 2
+    for worktree in worktrees:
+        assert worktree.is_dir()
+        assert (worktree / "greeting.py").read_text().strip() == (
+            "def greet(name):\n    return 'Hello, ' + name + '!'"
+        )
+
+
 def test_budget_exhausted_is_detected():
     result = run_model("noop", budget=Budget(move_budget=2, observation_budget=4000))
     assert result.success is False

tests/test_worktree.py

@@ -1,5 +1,8 @@
 from pathlib import Path
 
+import pytest
+
+import safe_mini.worktree as worktree_module
 from safe_mini.worktree import WorktreeProvisioner
 
 
@@ -29,3 +32,21 @@ def test_context_manager_cleans_up(tmp_path: Path):
         assert copy_path.exists()
         held = copy_path
     assert not held.exists()
+
+
+def test_provision_failure_removes_its_allocated_tempdir(tmp_path: Path, monkeypatch):
+    source = tmp_path / "source"
+    source.mkdir()
+    (source / "file.txt").write_text("original")
+
+    def fail_copy(*_args, **_kwargs):
+        raise OSError("simulated copy failure")
+
+    monkeypatch.setattr(worktree_module.shutil, "copytree", fail_copy)
+    provisioner = WorktreeProvisioner(source, tmp_root=tmp_path)
+
+    with pytest.raises(OSError, match="simulated copy failure"):
+        provisioner.provision()
+
+    assert provisioner.path is None
+    assert not list(tmp_path.glob("safe-mini-*"))