Freeze the adoption case script and record the r1 fail-stop
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+160
@@ -0,0 +1,160 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Adoption positive/negative cases through a repository's own scripts/agent-exec.
|
||||
|
||||
Usage: adopt_cases.py <adopted-repo> [<dm-source>]
|
||||
Clones the committed HEAD of <adopted-repo> into a temporary directory,
|
||||
materializes deps/development-manager at its gitlink, points every profile at a
|
||||
fake CLI and runs: positive run+apply, missing config, bad receipt, out-of-scope
|
||||
path and unknown side effect. Prints one line per case and a verdict.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
SRC = Path(sys.argv[1]).resolve()
|
||||
DM_SOURCE = sys.argv[2] if len(sys.argv) > 2 else str(Path.home() / "projects/Development-Manager")
|
||||
TERMINAL_OK = json.dumps({"terminal_status": "completed", "summary": "done", "changed_files": ["adoption-probe.txt"], "evidence": ["wrote probe"], "scope_expansion": None}, separators=(",", ":"))
|
||||
HELP = "--print\n--output-format\nstream-json\n--force\n--trust\n--workspace\n--sandbox\n--model\n-p\n"
|
||||
ENVELOPE = 'printf \'%s\\n\' "{\\"type\\":\\"result\\",\\"subtype\\":\\"success\\",\\"is_error\\":false,\\"result\\":$(python3 -c \'import json,sys;print(json.dumps(sys.stdin.read()))\' < "$R")}"'
|
||||
|
||||
|
||||
def sh(*args, cwd=None, env=None, check=True):
|
||||
return subprocess.run(list(args), cwd=cwd, env=env, capture_output=True, text=True, check=check, timeout=900)
|
||||
|
||||
|
||||
def fake_cli(bindir: Path, body: str) -> None:
|
||||
for name in ("fake-cli",):
|
||||
p = bindir / name
|
||||
p.write_text(
|
||||
"#!/bin/sh\n"
|
||||
'case "$1" in --version|-v|-V) echo 0.0.1-test; exit 0;; --help|-h) cat <<\'HELP\'\n' + HELP + "HELP\nexit 0;; esac\n"
|
||||
'R=$(mktemp)\n' + body + "\nrc=$?\n" + ENVELOPE + '\nrm -f "$R"\nexit $rc\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
p.chmod(p.stat().st_mode | stat.S_IEXEC)
|
||||
|
||||
|
||||
def prepare(root: Path, drop_config=False) -> Path:
|
||||
repo = root / "repo"
|
||||
sh("git", "clone", "-q", "--no-recurse-submodules", str(SRC), str(repo))
|
||||
sh("git", "-C", str(repo), "config", "user.email", "adopt@example.invalid")
|
||||
sh("git", "-C", str(repo), "config", "user.name", "adopt")
|
||||
pin = sh("git", "-C", str(repo), "ls-tree", "HEAD", "deps/development-manager").stdout.split()
|
||||
if len(pin) >= 3 and pin[0] == "160000":
|
||||
dm = repo / "deps/development-manager"
|
||||
if dm.exists():
|
||||
shutil.rmtree(dm)
|
||||
sh("git", "clone", "-q", "--no-checkout", DM_SOURCE, str(dm))
|
||||
sh("git", "-C", str(dm), "checkout", "-q", "--detach", pin[2])
|
||||
elif not (repo / "cmd/dm-server").is_dir():
|
||||
raise SystemExit("deps/development-manager is not a gitlink at HEAD")
|
||||
cfg = repo / "AGENT-EXECUTION.md"
|
||||
text = cfg.read_text(encoding="utf-8")
|
||||
text = re.sub(r"(?m)^(\s+executable:\s*).*$", r"\1fake-cli", text)
|
||||
text = re.sub(r"(?m)^(\s+adapter:\s*)\S+$", r"\1cursor", text)
|
||||
cfg.write_text(text, encoding="utf-8")
|
||||
if drop_config:
|
||||
cfg.unlink()
|
||||
sh("git", "-C", str(repo), "add", "-A", "--", "AGENT-EXECUTION.md")
|
||||
sh("git", "-C", str(repo), "commit", "-q", "--no-verify", "-m", "adoption case fixture")
|
||||
sh("git", "-C", str(repo), "checkout", "-q", "-b", "feature/adoption-case")
|
||||
return repo
|
||||
|
||||
|
||||
def unit(root: Path) -> Path:
|
||||
p = root / "unit.json"
|
||||
p.write_text(json.dumps({"unit_id": "adoption-case", "goal": "Create adoption-probe.txt", "allowed_paths": ["adoption-probe.txt"], "verification": ["test -f adoption-probe.txt"], "constraints": ["leave uncommitted"], "network": False, "external_cost": False, "timeout_seconds": 120}))
|
||||
return p
|
||||
|
||||
|
||||
def run(repo: Path, root: Path, body: str):
|
||||
bindir = root / "bin"
|
||||
bindir.mkdir(exist_ok=True)
|
||||
fake_cli(bindir, body)
|
||||
rec = root / "receipt.json"
|
||||
env = dict(os.environ, PATH=f"{bindir}:{os.environ['PATH']}", AGENT_EXEC_CACHE=str(Path.home() / ".cache/adoption-cases"))
|
||||
p = sh(str(repo / "scripts/agent-exec"), "run", "--repo", str(repo), "--role", "implementation", "--unit", str(unit(root)), "--receipt", str(rec), cwd=repo, env=env, check=False)
|
||||
d = json.loads(rec.read_text()) if rec.exists() else {}
|
||||
return p, d, rec, env
|
||||
|
||||
|
||||
WRITE_OK = 'echo probe > adoption-probe.txt; printf "%s" \'' + TERMINAL_OK + '\' > "$R"'
|
||||
results = []
|
||||
|
||||
|
||||
def case(name, fn):
|
||||
root = Path(tempfile.mkdtemp(prefix=f"adopt-{name}-"))
|
||||
try:
|
||||
ok, detail = fn(root)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
ok, detail = False, f"error: {exc}"
|
||||
finally:
|
||||
shutil.rmtree(root, ignore_errors=True)
|
||||
results.append(ok)
|
||||
print(f"{'PASS' if ok else 'FAIL'} {name}: {detail}")
|
||||
|
||||
|
||||
def positive(root):
|
||||
repo = prepare(root)
|
||||
p, d, rec, env = run(repo, root, WRITE_OK)
|
||||
if d.get("terminal_status") != "success" or not d.get("transport"):
|
||||
return False, f"run exit={p.returncode} terminal={d.get('terminal_status')} reason={d.get('reason')}"
|
||||
before = sh("git", "-C", str(repo), "rev-parse", "HEAD").stdout.strip()
|
||||
a = sh(str(repo / "scripts/agent-exec"), "apply", "--repo", str(repo), "--receipt", str(rec), cwd=repo, env=env, check=False)
|
||||
after = sh("git", "-C", str(repo), "rev-parse", "HEAD").stdout.strip()
|
||||
ok = a.returncode == 0 and after == d["transport"]["commit"] and before != after
|
||||
return ok, f"run exit={p.returncode} terminal=success apply exit={a.returncode} head={after[:12]}"
|
||||
|
||||
|
||||
def missing_config(root):
|
||||
repo = prepare(root, drop_config=True)
|
||||
p, d, _, env = run(repo, root, WRITE_OK)
|
||||
v = sh(str(repo / "scripts/agent-exec"), "validate", "--repo", str(repo), cwd=repo, env=env, check=False)
|
||||
ok = p.returncode != 0 and d.get("terminal_status") == "invalid-config" and not d.get("transport") and v.returncode != 0
|
||||
return ok, f"run exit={p.returncode} terminal={d.get('terminal_status')} validate exit={v.returncode}"
|
||||
|
||||
|
||||
def bad_receipt(root):
|
||||
repo = prepare(root)
|
||||
p, d, rec, env = run(repo, root, WRITE_OK)
|
||||
if d.get("terminal_status") != "success":
|
||||
return False, f"setup run terminal={d.get('terminal_status')}"
|
||||
d["changed_paths"] = sorted(set((d.get("changed_paths") or []) + ["src/injected.txt"]))
|
||||
rec.write_text(json.dumps(d))
|
||||
before = sh("git", "-C", str(repo), "rev-parse", "HEAD").stdout.strip()
|
||||
a = sh(str(repo / "scripts/agent-exec"), "apply", "--repo", str(repo), "--receipt", str(rec), cwd=repo, env=env, check=False)
|
||||
after = sh("git", "-C", str(repo), "rev-parse", "HEAD").stdout.strip()
|
||||
ok = a.returncode != 0 and before == after
|
||||
return ok, f"apply exit={a.returncode} head_unchanged={before == after} msg={(a.stderr or a.stdout).strip()[:100]}"
|
||||
|
||||
|
||||
def out_of_scope(root):
|
||||
repo = prepare(root)
|
||||
p, d, _, _ = run(repo, root, 'echo x > outside-scope.txt; ' + WRITE_OK)
|
||||
ok = p.returncode != 0 and d.get("terminal_status") == "scope-expansion" and not d.get("transport")
|
||||
return ok, f"run exit={p.returncode} terminal={d.get('terminal_status')} transport={bool(d.get('transport'))}"
|
||||
|
||||
|
||||
def unknown_side_effect(root):
|
||||
repo = prepare(root)
|
||||
p, d, _, _ = run(repo, root, 'echo probe > adoption-probe.txt; printf "done, no record" > "$R"')
|
||||
ok = p.returncode != 0 and d.get("terminal_status") == "failed-unknown" and not d.get("transport")
|
||||
return ok, f"run exit={p.returncode} terminal={d.get('terminal_status')} transport={bool(d.get('transport'))}"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
head = sh("git", "-C", str(SRC), "rev-parse", "HEAD").stdout.strip()
|
||||
print(f"[object] {SRC.name} HEAD={head}")
|
||||
case("positive-run-apply", positive)
|
||||
case("missing-config", missing_config)
|
||||
case("bad-receipt", bad_receipt)
|
||||
case("out-of-scope-path", out_of_scope)
|
||||
case("unknown-side-effect", unknown_side_effect)
|
||||
print(f"[verdict] {'ALL PASS' if all(results) else 'FAILURES'} cases={len(results)}")
|
||||
sys.exit(0 if all(results) else 1)
|
||||
Reference in New Issue
Block a user