Skip to content

Sandbox Execution Fabric

Disposable, isolated execution environments using Git worktrees.

dark_factory.sandbox.base.Sandbox

Bases: ABC

Abstract execution environment for running code agents and verification gates.

Source code in dark_factory/sandbox/base.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class Sandbox(ABC):
    """Abstract execution environment for running code agents and verification gates."""

    @abstractmethod
    def create(self, base_rev: str = "HEAD") -> None:
        """Initialize and isolate the sandbox at the specified revision."""

    @abstractmethod
    def execute(
        self,
        argv: list[str],
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        timeout: int = 300,
        step_id: str = "exec",
    ) -> StepExecution:
        """Execute a structured command inside the sandbox."""

    @abstractmethod
    def read_file(self, rel_path: str) -> bytes:
        """Read file contents relative to the sandbox root."""

    @abstractmethod
    def write_file(self, rel_path: str, content: bytes) -> None:
        """Write file contents relative to the sandbox root, creating parents if needed."""

    @abstractmethod
    def get_diff(self) -> str:
        """Extract a clean unified git diff of all modifications made in the sandbox."""

    @abstractmethod
    def restore_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
        """Restore specified paths to their baseline state at the given revision.

        `paths` entries are plain sandbox-relative paths OR git pathspec-magic strings such as
        `:(glob)**/conftest.py` (`VerificationRunner.detect_default_gate_paths()` emits both forms to
        protect files at any directory depth). Implementations MUST honor `:(glob)` magic, not just
        literal paths, or gate protection silently stops covering nested files.

        Returns the list of paths that were actually modified and restored.
        """

    def restore_runner_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
        """Restore runner import paths, including ignored shadows; fail on incomplete restoration.

        Drivers with ignored files must override this compatibility fallback.
        """
        return self.restore_paths(paths, rev=rev)

    @abstractmethod
    def destroy(self) -> None:
        """Tear down and destroy the sandbox idempotently."""

    def __enter__(self) -> Sandbox:
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self.destroy()

Methods:

create(base_rev='HEAD') abstractmethod

Initialize and isolate the sandbox at the specified revision.

Source code in dark_factory/sandbox/base.py 
13
14
15
@abstractmethod
def create(self, base_rev: str = "HEAD") -> None:
    """Initialize and isolate the sandbox at the specified revision."""

destroy() abstractmethod

Tear down and destroy the sandbox idempotently.

Source code in dark_factory/sandbox/base.py 
59
60
61
@abstractmethod
def destroy(self) -> None:
    """Tear down and destroy the sandbox idempotently."""

execute(argv, cwd=None, env=None, timeout=300, step_id='exec') abstractmethod

Execute a structured command inside the sandbox.

Source code in dark_factory/sandbox/base.py 
17
18
19
20
21
22
23
24
25
26
@abstractmethod
def execute(
    self,
    argv: list[str],
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    timeout: int = 300,
    step_id: str = "exec",
) -> StepExecution:
    """Execute a structured command inside the sandbox."""

get_diff() abstractmethod

Extract a clean unified git diff of all modifications made in the sandbox.

Source code in dark_factory/sandbox/base.py 
36
37
38
@abstractmethod
def get_diff(self) -> str:
    """Extract a clean unified git diff of all modifications made in the sandbox."""

read_file(rel_path) abstractmethod

Read file contents relative to the sandbox root.

Source code in dark_factory/sandbox/base.py 
28
29
30
@abstractmethod
def read_file(self, rel_path: str) -> bytes:
    """Read file contents relative to the sandbox root."""

restore_paths(paths, rev=None) abstractmethod

Restore specified paths to their baseline state at the given revision.

paths entries are plain sandbox-relative paths OR git pathspec-magic strings such as :(glob)**/conftest.py (VerificationRunner.detect_default_gate_paths() emits both forms to protect files at any directory depth). Implementations MUST honor :(glob) magic, not just literal paths, or gate protection silently stops covering nested files.

Returns the list of paths that were actually modified and restored.

Source code in dark_factory/sandbox/base.py 
40
41
42
43
44
45
46
47
48
49
50
@abstractmethod
def restore_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
    """Restore specified paths to their baseline state at the given revision.

    `paths` entries are plain sandbox-relative paths OR git pathspec-magic strings such as
    `:(glob)**/conftest.py` (`VerificationRunner.detect_default_gate_paths()` emits both forms to
    protect files at any directory depth). Implementations MUST honor `:(glob)` magic, not just
    literal paths, or gate protection silently stops covering nested files.

    Returns the list of paths that were actually modified and restored.
    """

restore_runner_paths(paths, rev=None)

Restore runner import paths, including ignored shadows; fail on incomplete restoration.

Drivers with ignored files must override this compatibility fallback.

Source code in dark_factory/sandbox/base.py 
52
53
54
55
56
57
def restore_runner_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
    """Restore runner import paths, including ignored shadows; fail on incomplete restoration.

    Drivers with ignored files must override this compatibility fallback.
    """
    return self.restore_paths(paths, rev=rev)

write_file(rel_path, content) abstractmethod

Write file contents relative to the sandbox root, creating parents if needed.

Source code in dark_factory/sandbox/base.py 
32
33
34
@abstractmethod
def write_file(self, rel_path: str, content: bytes) -> None:
    """Write file contents relative to the sandbox root, creating parents if needed."""

dark_factory.sandbox.worktree.GitWorktreeSandbox

Bases: Sandbox

A lightweight, isolated sandbox using git worktrees.

Source code in dark_factory/sandbox/worktree.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
class GitWorktreeSandbox(Sandbox):
    """A lightweight, isolated sandbox using git worktrees."""

    def __init__(
        self,
        repo_path: str | Path,
        sandbox_id: str,
        base_dir: str | Path | None = None,
    ) -> None:
        self.repo_path = Path(repo_path).resolve()
        self.sandbox_id = sandbox_id
        if base_dir is None:
            self.sandbox_dir = (self.repo_path / ".factory" / "sandboxes" / sandbox_id).resolve()
        else:
            self.sandbox_dir = (Path(base_dir).resolve() / sandbox_id).resolve()
        self._created = False
        self._base_sha: str | None = None

    @property
    def path(self) -> Path:
        return self.sandbox_dir

    def create(self, base_rev: str = "HEAD") -> None:
        """Create an isolated git worktree at the given revision."""
        if self._created:
            return

        if not (self.repo_path / ".git").exists() and not self._is_git_repo():
            raise WorktreeCreationError(f"Target path is not a git repository: {self.repo_path}")

        # Resolve commit SHA
        try:
            rev_res = subprocess.run(
                ["git", "rev-parse", base_rev],
                cwd=self.repo_path,
                capture_output=True,
                text=True,
                errors="replace",
                check=True,
            )
            self._base_sha = rev_res.stdout.strip()
        except subprocess.CalledProcessError as e:
            raise WorktreeCreationError(f"Failed to resolve revision '{base_rev}': {e.stderr}") from e

        # Ensure parent directory exists
        self.sandbox_dir.parent.mkdir(parents=True, exist_ok=True)

        # Create worktree
        cmd = ["git", "worktree", "add", "--detach", str(self.sandbox_dir), self._base_sha]
        try:
            subprocess.run(
                cmd,
                cwd=self.repo_path,
                capture_output=True,
                text=True,
                errors="replace",
                check=True,
            )
            self._created = True
        except subprocess.CalledProcessError as e:
            raise WorktreeCreationError(f"Failed to create git worktree at {self.sandbox_dir}: {e.stderr}") from e

    def execute(
        self,
        argv: list[str],
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        timeout: int = 300,
        step_id: str = "exec",
    ) -> StepExecution:
        """Execute a command inside the sandbox."""
        if not self._created:
            raise SandboxError("Sandbox has not been created yet.")

        target_cwd = self.sandbox_dir if cwd is None else (self.sandbox_dir / cwd).resolve()
        self._check_path_within_sandbox(target_cwd)

        # Build clean environment
        exec_env = os.environ.copy()
        # Strip git and python env vars that might leak from parent/outer runtime
        for key in ["GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "VIRTUAL_ENV", "PYTHONPATH"]:
            exec_env.pop(key, None)
        if env:
            exec_env.update(env)

        start_time = time.monotonic()
        timed_out = False

        try:
            process = subprocess.Popen(
                argv,
                cwd=target_cwd,
                env=exec_env,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                errors="replace",
            )
            stdout, stderr = process.communicate(timeout=timeout)
            exit_code = process.returncode
        except subprocess.TimeoutExpired:
            process.kill()
            stdout, stderr = process.communicate()
            exit_code = -1
            timed_out = True
        except Exception as e:
            duration = time.monotonic() - start_time
            return StepExecution(
                step_id=step_id,
                exit_code=1,
                stdout="",
                stderr=f"Execution error: {e}",
                duration_sec=duration,
                timed_out=False,
            )

        duration = time.monotonic() - start_time
        return StepExecution(
            step_id=step_id,
            exit_code=exit_code,
            stdout=stdout or "",
            stderr=stderr or "",
            duration_sec=duration,
            timed_out=timed_out,
        )

    def read_file(self, rel_path: str) -> bytes:
        """Read a file relative to sandbox root."""
        if not self._created:
            raise SandboxError("Sandbox has not been created yet.")
        target_file = (self.sandbox_dir / rel_path).resolve()
        self._check_path_within_sandbox(target_file)
        if not target_file.is_file():
            raise FileNotFoundError(f"File not found in sandbox: {rel_path}")
        return target_file.read_bytes()

    def write_file(self, rel_path: str, content: bytes) -> None:
        """Write a file relative to sandbox root, creating parent directories."""
        if not self._created:
            raise SandboxError("Sandbox has not been created yet.")
        target_file = (self.sandbox_dir / rel_path).resolve()
        self._check_path_within_sandbox(target_file)
        self._check_not_git_metadata(target_file)
        target_file.parent.mkdir(parents=True, exist_ok=True)
        target_file.write_bytes(content)

    def get_diff(self) -> str:
        """Extract unified git diff of all modifications including untracked files."""
        if not self._created:
            raise SandboxError("Sandbox has not been created yet.")

        # Clean python bytecode and test caches if any were generated
        try:
            for pyc in self.sandbox_dir.rglob("*.pyc"):
                pyc.unlink(missing_ok=True)
            for cache_dir in list(self.sandbox_dir.rglob("__pycache__")):
                if cache_dir.is_dir():
                    shutil.rmtree(cache_dir, ignore_errors=True)
            for cache_dir in list(self.sandbox_dir.rglob(".pytest_cache")):
                if cache_dir.is_dir():
                    shutil.rmtree(cache_dir, ignore_errors=True)
        except Exception:
            pass

        # Stage untracked files intent-to-add so diff captures new files too
        self._run_diff_git(["git", "add", "-N", "."])

        # Independent of user git config (color.ui, diff.external, diff.noprefix, diff.renames, ...): the
        # stored patch must be plain, a/ b/ prefixed, rename-free and binary-safe so it re-applies cleanly.
        return self._run_diff_git(
            [
                "git",
                "diff",
                "HEAD",
                "--no-color",
                "--no-ext-diff",
                "--no-renames",
                "--binary",
                "--src-prefix=a/",
                "--dst-prefix=b/",
                "--",
                ".",
                ":(exclude)__pycache__",
                ":(exclude)*.pyc",
                ":(exclude).pytest_cache",
            ]
        )

    def _run_diff_git(self, argv: list[str]) -> str:
        proc = subprocess.run(argv, cwd=self.sandbox_dir, capture_output=True, text=True, errors="replace")
        if proc.returncode != 0:
            raise SandboxError(f"git diff extraction failed at '{' '.join(argv[:3])}': {proc.stderr.strip()}")
        return proc.stdout

    def restore_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
        """Restore specified paths to their baseline state at the given revision.

        Returns the list of paths that were actually modified and restored.
        """
        if not self._created:
            raise SandboxError("Sandbox has not been created yet.")
        if not paths:
            return []

        target_rev = rev or self._base_sha or "HEAD"
        restored: list[str] = []

        for p in paths:
            rel = p.lstrip("/")
            # 1. Check if git reports modifications, deletions, or staged changes for path
            diff_res = subprocess.run(
                ["git", "diff", target_rev, "--name-only", "--", rel],
                cwd=self.sandbox_dir,
                capture_output=True,
                text=True,
                errors="replace",
            )
            # 2. Check for untracked new files under path
            untracked_res = subprocess.run(
                ["git", "ls-files", "--others", "--exclude-standard", "--", rel],
                cwd=self.sandbox_dir,
                capture_output=True,
                text=True,
                errors="replace",
            )
            has_diff = bool(diff_res.stdout.strip())
            has_untracked = bool(untracked_res.stdout.strip())

            if has_diff or has_untracked:
                # Restore tracked files to target_rev
                subprocess.run(
                    ["git", "checkout", target_rev, "--", rel],
                    cwd=self.sandbox_dir,
                    capture_output=True,
                    text=True,
                    errors="replace",
                )
                # Clean untracked files
                subprocess.run(
                    ["git", "clean", "-fd", "--", rel],
                    cwd=self.sandbox_dir,
                    capture_output=True,
                    text=True,
                    errors="replace",
                )
                restored.append(rel)

        return restored

    def _runner_git(self, args: list[str]) -> str:
        return subprocess.run(
            ["git", *args], cwd=self.sandbox_dir, capture_output=True, text=True, errors="replace", check=True
        ).stdout.strip()

    def restore_runner_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
        if not self._created:
            raise SandboxError("Sandbox not created")

        target_rev = rev or self._base_sha or "HEAD"
        restored = []

        for p in paths:
            try:
                changed = self._runner_git(["diff", target_rev, "--name-only", "--", p])
                untracked = self._runner_git(["ls-files", "--others", "--", p])

                if not changed and not untracked:
                    continue

                if self._runner_git(["ls-files", "--cached", "--", p]):
                    self._runner_git(["restore", "--source=" + target_rev, "--staged", "--worktree", "--", p])

                self._runner_git(["clean", "-fdx", "--", p])

                changed = self._runner_git(["diff", target_rev, "--name-only", "--", p])
                untracked = self._runner_git(["ls-files", "--others", "--", p])

                if changed or untracked:
                    raise SandboxError(f"Runner restoration incomplete: {p}")

                restored.append(p)

            except subprocess.CalledProcessError as exc:
                raise SandboxError(f"Runner restoration failed: {exc.stderr}") from exc

        return restored

    def destroy(self) -> None:
        """Destroy the worktree and clean up files idempotently."""
        if not self._created and not self.sandbox_dir.exists():
            return

        # Tell git to remove worktree
        try:
            subprocess.run(
                ["git", "worktree", "remove", "--force", str(self.sandbox_dir)],
                cwd=self.repo_path,
                capture_output=True,
                text=True,
                errors="replace",
                check=False,
            )
        except Exception:
            pass

        # Cleanup directory if still exists
        if self.sandbox_dir.exists():
            shutil.rmtree(self.sandbox_dir, ignore_errors=True)

        # Prune dead worktrees
        try:
            subprocess.run(
                ["git", "worktree", "prune"],
                cwd=self.repo_path,
                capture_output=True,
                text=True,
                errors="replace",
                check=False,
            )
        except Exception:
            pass

        self._created = False

    def _is_git_repo(self) -> bool:
        res = subprocess.run(
            ["git", "rev-parse", "--is-inside-work-tree"],
            cwd=self.repo_path,
            capture_output=True,
            text=True,
            errors="replace",
        )
        return res.returncode == 0

    def _check_not_git_metadata(self, path: Path) -> None:
        """Agents must never write `.git` (the worktree `gitdir:` pointer) or anything beneath it."""
        rel_parts = path.relative_to(self.sandbox_dir).parts
        if any(part.casefold() == ".git" for part in rel_parts):
            raise SandboxError(f"Writes to git metadata (.git) are forbidden: {path}")

    def _check_path_within_sandbox(self, path: Path) -> None:
        try:
            path.relative_to(self.sandbox_dir)
        except ValueError:
            raise SandboxError(f"Path traversal detected outside sandbox: {path}") from None

Methods:

create(base_rev='HEAD')

Create an isolated git worktree at the given revision.

Source code in dark_factory/sandbox/worktree.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def create(self, base_rev: str = "HEAD") -> None:
    """Create an isolated git worktree at the given revision."""
    if self._created:
        return

    if not (self.repo_path / ".git").exists() and not self._is_git_repo():
        raise WorktreeCreationError(f"Target path is not a git repository: {self.repo_path}")

    # Resolve commit SHA
    try:
        rev_res = subprocess.run(
            ["git", "rev-parse", base_rev],
            cwd=self.repo_path,
            capture_output=True,
            text=True,
            errors="replace",
            check=True,
        )
        self._base_sha = rev_res.stdout.strip()
    except subprocess.CalledProcessError as e:
        raise WorktreeCreationError(f"Failed to resolve revision '{base_rev}': {e.stderr}") from e

    # Ensure parent directory exists
    self.sandbox_dir.parent.mkdir(parents=True, exist_ok=True)

    # Create worktree
    cmd = ["git", "worktree", "add", "--detach", str(self.sandbox_dir), self._base_sha]
    try:
        subprocess.run(
            cmd,
            cwd=self.repo_path,
            capture_output=True,
            text=True,
            errors="replace",
            check=True,
        )
        self._created = True
    except subprocess.CalledProcessError as e:
        raise WorktreeCreationError(f"Failed to create git worktree at {self.sandbox_dir}: {e.stderr}") from e

destroy()

Destroy the worktree and clean up files idempotently.

Source code in dark_factory/sandbox/worktree.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def destroy(self) -> None:
    """Destroy the worktree and clean up files idempotently."""
    if not self._created and not self.sandbox_dir.exists():
        return

    # Tell git to remove worktree
    try:
        subprocess.run(
            ["git", "worktree", "remove", "--force", str(self.sandbox_dir)],
            cwd=self.repo_path,
            capture_output=True,
            text=True,
            errors="replace",
            check=False,
        )
    except Exception:
        pass

    # Cleanup directory if still exists
    if self.sandbox_dir.exists():
        shutil.rmtree(self.sandbox_dir, ignore_errors=True)

    # Prune dead worktrees
    try:
        subprocess.run(
            ["git", "worktree", "prune"],
            cwd=self.repo_path,
            capture_output=True,
            text=True,
            errors="replace",
            check=False,
        )
    except Exception:
        pass

    self._created = False

execute(argv, cwd=None, env=None, timeout=300, step_id='exec')

Execute a command inside the sandbox.

Source code in dark_factory/sandbox/worktree.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def execute(
    self,
    argv: list[str],
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    timeout: int = 300,
    step_id: str = "exec",
) -> StepExecution:
    """Execute a command inside the sandbox."""
    if not self._created:
        raise SandboxError("Sandbox has not been created yet.")

    target_cwd = self.sandbox_dir if cwd is None else (self.sandbox_dir / cwd).resolve()
    self._check_path_within_sandbox(target_cwd)

    # Build clean environment
    exec_env = os.environ.copy()
    # Strip git and python env vars that might leak from parent/outer runtime
    for key in ["GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "VIRTUAL_ENV", "PYTHONPATH"]:
        exec_env.pop(key, None)
    if env:
        exec_env.update(env)

    start_time = time.monotonic()
    timed_out = False

    try:
        process = subprocess.Popen(
            argv,
            cwd=target_cwd,
            env=exec_env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            errors="replace",
        )
        stdout, stderr = process.communicate(timeout=timeout)
        exit_code = process.returncode
    except subprocess.TimeoutExpired:
        process.kill()
        stdout, stderr = process.communicate()
        exit_code = -1
        timed_out = True
    except Exception as e:
        duration = time.monotonic() - start_time
        return StepExecution(
            step_id=step_id,
            exit_code=1,
            stdout="",
            stderr=f"Execution error: {e}",
            duration_sec=duration,
            timed_out=False,
        )

    duration = time.monotonic() - start_time
    return StepExecution(
        step_id=step_id,
        exit_code=exit_code,
        stdout=stdout or "",
        stderr=stderr or "",
        duration_sec=duration,
        timed_out=timed_out,
    )

get_diff()

Extract unified git diff of all modifications including untracked files.

Source code in dark_factory/sandbox/worktree.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def get_diff(self) -> str:
    """Extract unified git diff of all modifications including untracked files."""
    if not self._created:
        raise SandboxError("Sandbox has not been created yet.")

    # Clean python bytecode and test caches if any were generated
    try:
        for pyc in self.sandbox_dir.rglob("*.pyc"):
            pyc.unlink(missing_ok=True)
        for cache_dir in list(self.sandbox_dir.rglob("__pycache__")):
            if cache_dir.is_dir():
                shutil.rmtree(cache_dir, ignore_errors=True)
        for cache_dir in list(self.sandbox_dir.rglob(".pytest_cache")):
            if cache_dir.is_dir():
                shutil.rmtree(cache_dir, ignore_errors=True)
    except Exception:
        pass

    # Stage untracked files intent-to-add so diff captures new files too
    self._run_diff_git(["git", "add", "-N", "."])

    # Independent of user git config (color.ui, diff.external, diff.noprefix, diff.renames, ...): the
    # stored patch must be plain, a/ b/ prefixed, rename-free and binary-safe so it re-applies cleanly.
    return self._run_diff_git(
        [
            "git",
            "diff",
            "HEAD",
            "--no-color",
            "--no-ext-diff",
            "--no-renames",
            "--binary",
            "--src-prefix=a/",
            "--dst-prefix=b/",
            "--",
            ".",
            ":(exclude)__pycache__",
            ":(exclude)*.pyc",
            ":(exclude).pytest_cache",
        ]
    )

read_file(rel_path)

Read a file relative to sandbox root.

Source code in dark_factory/sandbox/worktree.py 
145
146
147
148
149
150
151
152
153
def read_file(self, rel_path: str) -> bytes:
    """Read a file relative to sandbox root."""
    if not self._created:
        raise SandboxError("Sandbox has not been created yet.")
    target_file = (self.sandbox_dir / rel_path).resolve()
    self._check_path_within_sandbox(target_file)
    if not target_file.is_file():
        raise FileNotFoundError(f"File not found in sandbox: {rel_path}")
    return target_file.read_bytes()

restore_paths(paths, rev=None)

Restore specified paths to their baseline state at the given revision.

Returns the list of paths that were actually modified and restored.

Source code in dark_factory/sandbox/worktree.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def restore_paths(self, paths: list[str], rev: str | None = None) -> list[str]:
    """Restore specified paths to their baseline state at the given revision.

    Returns the list of paths that were actually modified and restored.
    """
    if not self._created:
        raise SandboxError("Sandbox has not been created yet.")
    if not paths:
        return []

    target_rev = rev or self._base_sha or "HEAD"
    restored: list[str] = []

    for p in paths:
        rel = p.lstrip("/")
        # 1. Check if git reports modifications, deletions, or staged changes for path
        diff_res = subprocess.run(
            ["git", "diff", target_rev, "--name-only", "--", rel],
            cwd=self.sandbox_dir,
            capture_output=True,
            text=True,
            errors="replace",
        )
        # 2. Check for untracked new files under path
        untracked_res = subprocess.run(
            ["git", "ls-files", "--others", "--exclude-standard", "--", rel],
            cwd=self.sandbox_dir,
            capture_output=True,
            text=True,
            errors="replace",
        )
        has_diff = bool(diff_res.stdout.strip())
        has_untracked = bool(untracked_res.stdout.strip())

        if has_diff or has_untracked:
            # Restore tracked files to target_rev
            subprocess.run(
                ["git", "checkout", target_rev, "--", rel],
                cwd=self.sandbox_dir,
                capture_output=True,
                text=True,
                errors="replace",
            )
            # Clean untracked files
            subprocess.run(
                ["git", "clean", "-fd", "--", rel],
                cwd=self.sandbox_dir,
                capture_output=True,
                text=True,
                errors="replace",
            )
            restored.append(rel)

    return restored

write_file(rel_path, content)

Write a file relative to sandbox root, creating parent directories.

Source code in dark_factory/sandbox/worktree.py 
155
156
157
158
159
160
161
162
163
def write_file(self, rel_path: str, content: bytes) -> None:
    """Write a file relative to sandbox root, creating parent directories."""
    if not self._created:
        raise SandboxError("Sandbox has not been created yet.")
    target_file = (self.sandbox_dir / rel_path).resolve()
    self._check_path_within_sandbox(target_file)
    self._check_not_git_metadata(target_file)
    target_file.parent.mkdir(parents=True, exist_ok=True)
    target_file.write_bytes(content)