| """Full git checkout of a SWE-bench instance repo at base_commit.""" | |
| from __future__ import annotations | |
| import os | |
| import shutil | |
| import subprocess | |
| from pathlib import Path | |
| from deepagents_rl.observability.logging import call, error, result, span, truncate | |
| def repo_cache_root() -> Path: | |
| root = Path(os.getenv("SWE_REPO_CACHE", "")) if os.getenv("SWE_REPO_CACHE") else None | |
| if root is None: | |
| root = Path(__file__).resolve().parents[3] / "data" / "repos" | |
| root.mkdir(parents=True, exist_ok=True) | |
| return root | |
| def _run_git( | |
| args: list[str], | |
| *, | |
| cwd: str | Path | None = None, | |
| check: bool = True, | |
| timeout: int = 600, | |
| ) -> subprocess.CompletedProcess[str]: | |
| cmd = ["git", *args] | |
| call("git", " ".join(cmd), cwd=str(cwd) if cwd else None, timeout_s=timeout) | |
| try: | |
| proc = subprocess.run( | |
| cmd, | |
| cwd=str(cwd) if cwd else None, | |
| check=False, | |
| capture_output=True, | |
| text=True, | |
| timeout=timeout, | |
| ) | |
| except subprocess.TimeoutExpired as e: | |
| error("git", f"timeout after {timeout}s", cmd=cmd) | |
| raise e | |
| out = ((proc.stdout or "") + (proc.stderr or "")).strip() | |
| result( | |
| "git", | |
| f"exit={proc.returncode}", | |
| exit_code=proc.returncode, | |
| output=truncate(out, 600), | |
| ) | |
| if check and proc.returncode != 0: | |
| raise subprocess.CalledProcessError(proc.returncode, cmd, proc.stdout, proc.stderr) | |
| return proc | |
| def clone_or_fetch(repo: str, cache_dir: Path | None = None) -> Path: | |
| """Mirror-clone (once) github.com/{repo} into the local cache.""" | |
| cache = cache_dir or repo_cache_root() | |
| mirror = cache / f"{repo.replace('/', '__')}.git" | |
| url = f"https://github.com/{repo}.git" | |
| with span("repo.clone_or_fetch", repo=repo, mirror=str(mirror)): | |
| if mirror.exists(): | |
| call("repo.fetch", f"fetch existing mirror {mirror}") | |
| _run_git( | |
| ["fetch", "--all", "--tags", "--prune"], | |
| cwd=mirror, | |
| check=False, | |
| timeout=600, | |
| ) | |
| return mirror | |
| mirror.parent.mkdir(parents=True, exist_ok=True) | |
| call("repo.mirror_clone", f"clone --mirror {url}") | |
| _run_git(["clone", "--mirror", url, str(mirror)], timeout=1800) | |
| return mirror | |
| def materialize_repo_at_commit( | |
| *, | |
| repo: str, | |
| base_commit: str, | |
| dest: Path, | |
| cache_dir: Path | None = None, | |
| ) -> Path: | |
| """ | |
| Create a full working tree at `dest` checked out to `base_commit`. | |
| This is the agent sandbox — not a sparse file dump. | |
| """ | |
| with span( | |
| "repo.materialize", | |
| f"{repo}@{base_commit[:12]}", | |
| repo=repo, | |
| base_commit=base_commit, | |
| dest=str(dest), | |
| ): | |
| if dest.exists(): | |
| call("repo.rmtree", str(dest)) | |
| shutil.rmtree(dest, ignore_errors=True) | |
| dest.mkdir(parents=True, exist_ok=True) | |
| mirror = clone_or_fetch(repo, cache_dir=cache_dir) | |
| call("repo.workdir_clone", f"from {mirror} -> {dest}") | |
| _run_git(["clone", str(mirror), str(dest)], timeout=600) | |
| call("repo.checkout", base_commit) | |
| _run_git(["checkout", "--force", base_commit], cwd=dest, timeout=120) | |
| _run_git(["reset", "--hard", "HEAD"], cwd=dest, timeout=60) | |
| _run_git(["clean", "-fdx"], cwd=dest, timeout=60) | |
| head = _run_git(["rev-parse", "HEAD"], cwd=dest, timeout=30) | |
| result( | |
| "repo.materialize", | |
| "sandbox ready", | |
| head=(head.stdout or "").strip(), | |
| dest=str(dest), | |
| ) | |
| return dest | |
| def git_diff_patch(repo_dir: Path) -> str: | |
| """Unified diff of all agent edits (tracked + untracked as new files).""" | |
| with span("repo.git_diff_patch", repo_dir=str(repo_dir)): | |
| _run_git(["add", "-A"], cwd=repo_dir, timeout=60) | |
| proc = _run_git(["diff", "--cached", "--binary"], cwd=repo_dir, timeout=60) | |
| _run_git(["reset", "HEAD"], cwd=repo_dir, check=False, timeout=60) | |
| patch = proc.stdout or "" | |
| result( | |
| "repo.git_diff_patch", | |
| f"patch_bytes={len(patch.encode('utf-8'))}", | |
| patch_preview=truncate(patch, 500), | |
| empty=not bool(patch.strip()), | |
| ) | |
| return patch | |