91 lines
4.7 KiB
Python
91 lines
4.7 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):
|
|
gates = root / ".local/group-movement-gates"
|
|
candidates = sorted(gates.glob("*.py")) + helpers
|
|
helper_bytes = [path.read_bytes() for path in candidates]
|
|
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
|
|
for run in runs:
|
|
folder = gates / run
|
|
need(folder.is_dir(), "run directory missing: " + run)
|
|
paths = list(folder.iterdir())
|
|
need(all(path.is_file() and not path.is_symlink() for path in paths)
|
|
and {path.name for path in paths} == RAW_FILES, "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
|
|
manifest = {"schema": 1, "kind": "multiselect-cargo-gates", "runs": specs,
|
|
"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, default=HERE.parents[3], help="checkout holding .local/group-movement-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("--output", type=Path, default=HERE, help="package directory; existing evidence is never overwritten")
|
|
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")
|
|
payload, manifest, summary = collect(args.root.resolve(), args.runs, args.helper)
|
|
if args.check_only:
|
|
print("PASS raw consistency; no package written:", summary)
|
|
return
|
|
out = args.output.resolve()
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
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()
|