sots-re/verify/results/research-completion-abi/run-9daf5c3b75547271d5c3b4ed/capture.py
2026-09-10 01:07:05 -04:00

163 lines
7.1 KiB
Python

#!/usr/bin/env python3
"""Fresh, session-bound static capture required by two 2026-09-10 Astra decisions."""
import argparse
import hashlib
import json
import subprocess
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path("/home/alex/sots-re")
SESSION = "run-9daf5c3b75547271d5c3b4ed"
STALE_SESSION = "run-ee78b8773688ca09f8046e21"
REL_OUT = Path("verify/results/research-completion-abi") / SESSION
OUT = ROOT / REL_OUT
BINARY = ROOT / "dumps/sots.exe"
TOOL = Path("/usr/bin/objdump")
CAMPAIGN = ROOT / "tools/campaign.py"
ENGINE_TREE = Path("/tmp/opencode/sots-final-research-engine")
RE_TREE = Path("/tmp/opencode/sots-final-research-re")
EXPECTED_BINARY = "970b7de729956a53094c7eb98aba4270aee98e2fed5daf0d39e290013c90c841"
EXPECTED_TOOL = "1eaaef2e7f57c4c7f69115c495e2466f5a8c8e5f3bc42221d092382f30f9d4cd"
# Exact historical windows remain controls. Each companion widened stop was already shown
# to add rows without changing the exact rows, except nested-copy's known-bad historical stop.
DISASSEMBLY = (
("turn-get-create-exact", 0x00885380, 0x0088544A, "complete-control"),
("turn-get-create-wide", 0x00885380, 0x0088544E, "boundary-audit"),
("turn-append-exact", 0x00884CB0, 0x00884D8F, "complete-control"),
("turn-append-wide", 0x00884CB0, 0x00884D93, "boundary-audit"),
("nested-dtor-exact", 0x00629580, 0x006295CA, "complete-control"),
("nested-dtor-wide", 0x00629580, 0x006295CE, "boundary-audit"),
("nested-copy-authority", 0x00779850, 0x00779930, "bounded-authority"),
("nested-copy-stop-a20", 0x00779850, 0x00779A20, "expected-truncated-negative"),
("nested-copy-stop-a21", 0x00779850, 0x00779A21, "predicted-complete-je"),
("nested-copy-stop-a23", 0x00779850, 0x00779A23, "predicted-complete-mov"),
("nested-copy-stop-a28", 0x00779850, 0x00779A28, "complete-call-boundary-control"),
)
RAW_BYTES = (
("nested-copy-boundary-bytes", 0x00779A1F, 0x00779A28),
)
def sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def file_identity(path: Path) -> dict:
data = path.read_bytes()
return {"path": str(path), "bytes": len(data), "sha256": sha(data)}
def validate_identity(session: str, output: Path, entries: list[str]) -> None:
if not session:
raise ValueError("missing session")
if session == STALE_SESSION or session != SESSION:
raise ValueError("stale or unexpected session")
if output.resolve() != OUT.resolve() or output.name != session:
raise ValueError("output location is not bound to session")
if sorted(entries) not in ([], ["capture.py"]):
raise FileExistsError("output location was already used")
def source_binding() -> dict:
argv = [
"python3", str(CAMPAIGN), "--state-root", str(ROOT), "source-binding",
"research-completion-abi", "--engine-worktree", str(ENGINE_TREE),
"--re-worktree", str(RE_TREE),
]
proc = subprocess.run(argv, cwd=ROOT, check=True, capture_output=True, text=True)
return {"argv": argv, "value": json.loads(proc.stdout)}
def run_capture(name: str, argv: list[str], classification: str) -> dict:
proc = subprocess.run(argv, cwd=ROOT, check=False, capture_output=True)
stdout_path = OUT / f"{name}.stdout.txt"
stderr_path = OUT / f"{name}.stderr.txt"
stdout_path.write_bytes(proc.stdout)
stderr_path.write_bytes(proc.stderr)
return {
"name": name,
"classification": classification,
"argv": argv,
"cwd": str(ROOT),
"returncode": proc.returncode,
"stdout": {"path": str(stdout_path.relative_to(ROOT)), "bytes": len(proc.stdout), "sha256": sha(proc.stdout)},
"stderr": {"path": str(stderr_path.relative_to(ROOT)), "bytes": len(proc.stderr), "sha256": sha(proc.stderr)},
}
parser = argparse.ArgumentParser()
parser.add_argument("--session", required=True)
parser.add_argument("--output", required=True, type=Path)
args = parser.parse_args()
negative_preflight = []
for name, candidate_session, candidate_output, entries in (
("reject-missing-session", "", OUT, ["capture.py"]),
("reject-stale-session", STALE_SESSION, OUT, ["capture.py"]),
("reject-stale-output", SESSION, ROOT / "verify/results/research-completion-abi" / STALE_SESSION, ["capture.py"]),
("reject-reused-output", SESSION, OUT, ["capture.py", "prior-output.txt"]),
):
try:
validate_identity(candidate_session, candidate_output, entries)
except (ValueError, FileExistsError) as exc:
negative_preflight.append({"name": name, "status": "pass", "rejection": str(exc)})
else:
raise SystemExit(f"negative preflight unexpectedly accepted: {name}")
validate_identity(args.session, args.output, [p.name for p in OUT.iterdir()])
before = source_binding()
binary = file_identity(BINARY)
tool = file_identity(TOOL)
if binary["sha256"] != EXPECTED_BINARY or tool["sha256"] != EXPECTED_TOOL:
raise SystemExit("binary/tool identity drift")
records = []
for name, start, stop, classification in DISASSEMBLY:
records.append(run_capture(name, [
str(TOOL), "-D", "-Mintel", f"--start-address=0x{start:08x}",
f"--stop-address=0x{stop:08x}", str(BINARY),
], classification))
for name, start, stop in RAW_BYTES:
records.append(run_capture(name, [
str(TOOL), "-s", "-j", ".text", f"--start-address=0x{start:08x}",
f"--stop-address=0x{stop:08x}", str(BINARY),
], "raw-byte-boundary-control"))
after = source_binding()
checks = {
"explicit-session": args.session == SESSION,
"session-output-match": args.output.resolve() == OUT.resolve(),
"negative-preflight-all-rejected": all(x["status"] == "pass" for x in negative_preflight),
"binary-hash": binary["sha256"] == EXPECTED_BINARY,
"tool-hash": tool["sha256"] == EXPECTED_TOOL,
"source-stable": before["value"] == after["value"],
"all-return-zero": all(x["returncode"] == 0 for x in records),
"all-stdout-nonempty": all(x["stdout"]["bytes"] > 0 for x in records),
"all-stderr-empty": all(x["stderr"]["bytes"] == 0 for x in records),
}
manifest = {
"schema": "sots-abi-static-capture/1",
"session": SESSION,
"actor": "research-abi-analyst",
"role": "analyst",
"model": "openai/gpt-5.6-sol",
"timestamp": datetime.now(timezone.utc).isoformat(),
"decisions": ["d-fd5aff1eaf78a8c15d96723c", "d-0bb927e63b915c87a58d4257"],
"scope": "Fresh read-only static capture; no game, allocator, constructor, destructor, exception, or RNG execution.",
"negative_preflight": negative_preflight,
"source_before": before,
"source_after": after,
"input": binary,
"tool": {**tool, "version": subprocess.run([str(TOOL), "--version"], check=True, capture_output=True, text=True).stdout.splitlines()[0]},
"commands": records,
"checks": [{"name": name, "status": "pass" if status else "fail"} for name, status in checks.items()],
"failed": [name for name, status in checks.items() if not status],
}
(OUT / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
print(json.dumps({"commands": len(records), "failed": manifest["failed"], "manifest": str((OUT / "manifest.json").relative_to(ROOT))}, indent=2))
raise SystemExit(bool(manifest["failed"]))