File size: 6,567 Bytes
d61821a | 1 2 3 4 5 6 7 8 9 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 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 | """Validate held-out task patches against clean archived Git snapshots."""
from __future__ import annotations
import argparse
from io import BytesIO
import json
from pathlib import Path
import shlex
import subprocess
import tarfile
import tempfile
import time
from typing import Any, Iterable
from agent_harness.specs import TaskSpec, load_tasks
def run(
arguments: list[str],
cwd: Path,
timeout: int = 900,
) -> dict[str, Any]:
started = time.monotonic()
result = subprocess.run(
arguments,
cwd=cwd,
check=False,
capture_output=True,
text=True,
timeout=timeout,
)
combined = result.stdout + result.stderr
return {
"command": shlex.join(arguments),
"exit_code": result.returncode,
"elapsed_seconds": time.monotonic() - started,
"output_tail": combined[-12000:],
}
def extract_snapshot(repository: Path, commit: str, destination: Path) -> None:
archive = subprocess.run(
["git", "archive", "--format=tar", commit],
cwd=repository,
check=True,
capture_output=True,
timeout=120,
).stdout
with tarfile.open(fileobj=BytesIO(archive), mode="r:") as handle:
for member in handle.getmembers():
path = Path(member.name)
if path.is_absolute() or ".." in path.parts:
raise RuntimeError(f"Unsafe git archive member: {member.name}")
handle.extractall(destination)
def apply_patch(checkout: Path, patch: Path) -> dict[str, Any]:
return run(
["git", "apply", "--whitespace=nowarn", str(patch.resolve())],
checkout,
timeout=120,
)
def run_commands(checkout: Path, commands: Iterable[str]) -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
for command in commands:
arguments = shlex.split(command)
if not arguments or arguments[0] != "go" or arguments[1:2] != ["test"]:
raise RuntimeError(f"Only explicit go test validation is allowed: {command}")
results.append(run(arguments, checkout))
return results
def all_pass(results: Iterable[dict[str, Any]]) -> bool:
values = list(results)
return bool(values) and all(item["exit_code"] == 0 for item in values)
def validate_task(root: Path, repository: Path, task: TaskSpec) -> dict[str, Any]:
if not task.gold_patch or not task.test_patch:
raise RuntimeError(f"{task.task_id} has no source/test patch pair")
source_patch = root / "tasks" / task.gold_patch
test_patch = root / "tasks" / task.test_patch
started = time.monotonic()
with tempfile.TemporaryDirectory(prefix=f"{task.task_id.lower()}-") as raw_checkout:
checkout = Path(raw_checkout)
extract_snapshot(repository, task.base_commit, checkout)
original_tests = run_commands(checkout, task.pass_to_pass_tests)
test_apply = apply_patch(checkout, test_patch)
test_only_results = (
run_commands(checkout, task.fail_to_pass_tests)
if test_apply["exit_code"] == 0
else []
)
source_apply = apply_patch(checkout, source_patch)
gold_results = (
run_commands(checkout, task.pass_to_pass_tests)
if source_apply["exit_code"] == 0
else []
)
original_pass = all_pass(original_tests)
test_exposes_bug = bool(test_only_results) and any(
item["exit_code"] != 0 for item in test_only_results
)
gold_pass = all_pass(gold_results)
valid = (
original_pass
and test_apply["exit_code"] == 0
and test_exposes_bug
and source_apply["exit_code"] == 0
and gold_pass
)
return {
"schema_version": 1,
"task_id": task.task_id,
"base_commit": task.base_commit,
"gold_commit": task.gold_commit,
"valid_end_to_end": valid,
"checks": {
"original_suite_passes": original_pass,
"test_patch_applies": test_apply["exit_code"] == 0,
"test_patch_exposes_bug": test_exposes_bug,
"source_patch_applies": source_apply["exit_code"] == 0,
"gold_suite_passes": gold_pass,
},
"original_tests": original_tests,
"test_patch_apply": test_apply,
"test_only_results": test_only_results,
"source_patch_apply": source_apply,
"gold_results": gold_results,
"elapsed_seconds": time.monotonic() - started,
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
parser.add_argument("--repository", type=Path, default=Path("data/repos/gitlab-runner"))
parser.add_argument("--task", action="append")
args = parser.parse_args()
root = args.root.resolve()
repository = args.repository.resolve()
selected = set(args.task or [])
tasks = [
task
for task_id, task in load_tasks(root).items()
if task_id.startswith("TASK_CR_") and (not selected or task_id in selected)
]
if not tasks:
raise SystemExit("No confirmatory tasks selected")
output_dir = root / "tasks" / "validation"
output_dir.mkdir(parents=True, exist_ok=True)
go_version = subprocess.run(
["go", "version"], check=True, capture_output=True, text=True, timeout=30
).stdout.strip()
summary: list[dict[str, Any]] = []
for task in tasks:
print(f"VALIDATING {task.task_id}", flush=True)
result = validate_task(root, repository, task)
result["go_version"] = go_version
(output_dir / f"{task.task_id}.json").write_text(
json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
summary.append(
{
"task_id": task.task_id,
"valid_end_to_end": result["valid_end_to_end"],
"elapsed_seconds": result["elapsed_seconds"],
"checks": result["checks"],
}
)
print(json.dumps(summary[-1], sort_keys=True), flush=True)
(output_dir / "summary.json").write_text(
json.dumps(
{
"schema_version": 1,
"go_version": go_version,
"tasks": summary,
"valid_task_ids": [item["task_id"] for item in summary if item["valid_end_to_end"]],
},
indent=2,
sort_keys=True,
)
+ "\n",
encoding="utf-8",
)
if __name__ == "__main__":
main()
|