111 lines
6.2 KiB
Python
111 lines
6.2 KiB
Python
"""Package explicitly selected closed gate captures without running Cargo."""
|
|
from pathlib import Path
|
|
import argparse
|
|
import json
|
|
import zipfile
|
|
from verify import RAW_FILES, pin, parsed, need, run_name, verify_data, verify_package
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
def collect(root, runs, helpers, comparators=()):
|
|
gates = root / ".local/scenario-group-gates"
|
|
candidates = [gates / name for name in ("gate.py", "source-proof.py")] + helpers
|
|
helper_bytes = [path.read_bytes() for path in candidates]
|
|
comparator_paths = [root / ".local/process-source-evidence-tools/verify.py", *comparators]
|
|
comparator_bytes = [path.read_bytes() for path in comparator_paths if path.is_file()]
|
|
payload, logical, specs = {}, {}, {}
|
|
def add(name, data):
|
|
stored = ("capsules/" + pin(data)["sha256"] + Path(name).suffix
|
|
if name.rsplit("/", 1)[1] in {"inputs.json", "inputs.zip"} else name)
|
|
need(stored not in payload or payload[stored] == data, "deduplication collision")
|
|
payload[stored] = data
|
|
logical[name] = stored
|
|
helper_catalog = {}
|
|
for name, data in zip(("gate.py", "source-proof.py"), helper_bytes):
|
|
helper = "helpers/" + pin(data)["sha256"] + ".py"
|
|
add(helper, data)
|
|
helper_catalog[name] = helper
|
|
for run in runs:
|
|
folder = gates / run
|
|
need(folder.is_dir(), "run directory missing: " + run)
|
|
paths = list(folder.iterdir())
|
|
names = {path.name for path in paths}
|
|
need(all(path.is_file() and not path.is_symlink() for path in paths)
|
|
and names in (RAW_FILES, RAW_FILES | {"source-proof.json"}), "exact closed raw files required: " + run)
|
|
raw = {path.name: path.read_bytes() for path in paths}
|
|
result = parsed(raw["result.json"])
|
|
need("finish" in result and result["inputsUnchanged"] is True, "unfinished/changed run refused: " + run)
|
|
matches = [data for data in helper_bytes if pin(data) == result["helper"]]
|
|
need(bool(matches), "exact historical helper unavailable for " + run + "; supply --helper PATH")
|
|
helper = "helpers/" + result["helper"]["sha256"] + ".py"
|
|
add(helper, matches[0])
|
|
for name, data in sorted(raw.items()):
|
|
add("cargo/" + run + "/" + name, data)
|
|
specs[run] = {key: result[key] for key in ("command", "head", "exitCode", "inputCount")}
|
|
specs[run]["helper"] = helper
|
|
if "source-proof.json" in raw:
|
|
proof = parsed(raw["source-proof.json"])
|
|
matches = [data for data in comparator_bytes if pin(data) == proof["comparator"]]
|
|
need(bool(matches), "exact source comparator unavailable for " + run + "; supply --comparator PATH")
|
|
comparator = "comparators/" + proof["comparator"]["sha256"] + ".py"
|
|
add(comparator, matches[0])
|
|
specs[run]["sourceProof"] = {"commit": proof["commit"], "pin": pin(raw["source-proof.json"]), "comparator": comparator}
|
|
manifest = {"schema": 1, "kind": "scenario-selection-cargo-gates", "runs": specs,
|
|
"captureHelpers": helper_catalog,
|
|
"logicalFiles": dict(sorted(logical.items())),
|
|
"members": {name: pin(data) for name, data in sorted(payload.items())}}
|
|
summary = verify_data(manifest, payload.__getitem__)
|
|
return payload, manifest, summary
|
|
|
|
|
|
def write_json(path, value):
|
|
with path.open("xb") as output:
|
|
output.write((json.dumps(value, indent=2) + "\n").encode("utf-8"))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", type=Path, required=True, help="checkout holding .local/scenario-group-gates")
|
|
parser.add_argument("--runs", nargs="+", required=True, help="explicit run names, including failures to retain")
|
|
parser.add_argument("--helper", type=Path, action="append", default=[], help="additional exact historical helper (repeatable)")
|
|
parser.add_argument("--comparator", type=Path, action="append", default=[], help="additional exact historical source comparator (repeatable)")
|
|
parser.add_argument("--output", type=Path, help="fresh, nonexistent package directory (required unless --check-only)")
|
|
parser.add_argument("--check-only", action="store_true", help="validate raw captures in memory without creating an archive")
|
|
args = parser.parse_args()
|
|
need(all(run_name(run) for run in args.runs) and len(args.runs) == len(set(args.runs)), "unique safe run names required")
|
|
need(args.check_only or args.output is not None, "--output is required unless --check-only")
|
|
if not args.check_only:
|
|
need(not args.output.exists(), "output must be a fresh directory: " + str(args.output))
|
|
payload, manifest, summary = collect(args.root.resolve(), args.runs, args.helper, args.comparator)
|
|
if args.check_only:
|
|
print("PASS raw consistency; no package written:", summary)
|
|
return
|
|
out = args.output.resolve()
|
|
out.mkdir(parents=True, exist_ok=False)
|
|
for name in ("evidence.zip", "manifest.json", "files.json"):
|
|
need(not (out / name).exists(), "existing package output refused: " + name)
|
|
for name in ("prepare.py", "verify.py", "README.md"):
|
|
data = (HERE / name).read_bytes()
|
|
destination = out / name
|
|
if destination.exists():
|
|
need(destination.read_bytes() == data, "different existing package tool: " + name)
|
|
else:
|
|
with destination.open("xb") as output:
|
|
output.write(data)
|
|
with zipfile.ZipFile(out / "evidence.zip", "x", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
|
|
for name, data in sorted(payload.items()):
|
|
info = zipfile.ZipInfo(name, date_time=(2026, 9, 23, 0, 0, 0))
|
|
info.compress_type = zipfile.ZIP_DEFLATED
|
|
info.external_attr = 0o100644 << 16
|
|
archive.writestr(info, data)
|
|
write_json(out / "manifest.json", manifest)
|
|
names = ("prepare.py", "verify.py", "README.md", "manifest.json", "evidence.zip")
|
|
write_json(out / "files.json", {name: pin((out / name).read_bytes()) for name in names})
|
|
need(verify_package(out) == summary, "written package differs from validated captures")
|
|
print("PASS package readback:", out, summary)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|