130 lines
7.1 KiB
Python
130 lines
7.1 KiB
Python
"""Package explicit completed native runs; helpers are pinned data, never executed."""
|
|
from pathlib import Path
|
|
import argparse
|
|
import json
|
|
import re
|
|
import zipfile
|
|
from verify import pin, parsed, need, norm, relative, safe, verify_data, verify_package
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
def collect(root, runs):
|
|
native = root / ".local/multiselect-native"
|
|
candidates = [path for path in native.iterdir() if path.is_file() and path.suffix in (".py", ".js")]
|
|
candidates.extend(sorted((root / ".local/group-movement-gates").glob("*.py")))
|
|
helpers = [(path.suffix, path.read_bytes()) for path in candidates]
|
|
payload, specs = {}, {}
|
|
def add(name, data):
|
|
need(safe(name) and (name not in payload or payload[name] == data), "member collision: " + name)
|
|
payload[name] = data
|
|
def tree(folder, prefix):
|
|
need(folder.is_dir(), "missing directory: " + str(folder))
|
|
for path in sorted(folder.rglob("*")):
|
|
need(not path.is_symlink(), "symlink evidence input")
|
|
if path.is_file():
|
|
add(prefix + path.relative_to(folder).as_posix(), path.read_bytes())
|
|
for run in runs:
|
|
folder = native / run
|
|
start = parsed((folder / "started.json").read_bytes())
|
|
end = parsed((folder / "finished.json").read_bytes())
|
|
need(end["exitCode"] == 0 and all(end[k] is True for k in
|
|
("guiUnchanged", "checkerUnchanged", "fixtureUnchanged")), "unfinished/changed native run: " + run)
|
|
generation = parsed((folder / "generation.json").read_bytes())
|
|
history = parsed((folder / "history-bytes.json").read_bytes())
|
|
tree(folder, "runs/" + run + "/raw/")
|
|
fixture_relative = relative(generation["command"][2], start["cwd"])
|
|
need(fixture_relative.startswith(".local/multiselect-native/"), "fixture outside native root")
|
|
tree(root / fixture_relative, "runs/" + run + "/fixture/")
|
|
spec = {"originalRun": norm(start["cwd"]) + "/.local/multiselect-native/" + run,
|
|
"originalFixture": norm(generation["command"][2]), "helpers": {}, "builds": {}}
|
|
for key, expected in [("helper", start["helper"]), ("capture", start["capture"]),
|
|
("snapshotHelper", start["snapshotHelper"]), ("history", history["helper"])]:
|
|
matches = [(suffix, data) for suffix, data in helpers if pin(data) == expected]
|
|
need(bool(matches), "exact historical helper unavailable: " + key)
|
|
suffix, data = matches[0]
|
|
member = "helpers/" + expected["sha256"] + suffix
|
|
add(member, data)
|
|
spec["helpers"][key] = member
|
|
for kind in ("gui", "checker"):
|
|
recorded = start[kind]
|
|
rel = relative(recorded["receipt"], start["cwd"])
|
|
need(rel.startswith(".local/group-movement-gates/") and rel.endswith("/result.json"), "unexpected build receipt location")
|
|
receipt = root / rel
|
|
data = receipt.read_bytes()
|
|
need(pin(data) == recorded["receiptPin"], "historical build receipt changed")
|
|
prefix = "builds/" + pin(data)["sha256"] + "/"
|
|
add(prefix + "result.json", data)
|
|
inputs_data = (receipt.parent / "inputs.json").read_bytes()
|
|
need(pin(inputs_data) == recorded["inputs"], "historical build input map changed")
|
|
add(prefix + "inputs.json", inputs_data)
|
|
spec["builds"][kind] = {"receipt": prefix + "result.json", "inputs": prefix + "inputs.json", "originalRelative": rel}
|
|
build_helper = parsed(data)["helper"]
|
|
matching_helpers = [(suffix, value) for suffix, value in helpers if pin(value) == build_helper]
|
|
need(bool(matching_helpers), "exact historical build helper unavailable")
|
|
suffix, helper_data = matching_helpers[0]
|
|
helper_member = "helpers/" + build_helper["sha256"] + suffix
|
|
add(helper_member, helper_data)
|
|
spec["builds"][kind]["helper"] = helper_member
|
|
if kind == "checker":
|
|
# Read only the pinned historical example, not today's checkout.
|
|
archive_path = receipt.parent / "inputs.zip"
|
|
need(pin(archive_path.read_bytes()) == parsed(data)["archive"], "checker input archive pin")
|
|
source_path = "crates/oa-core/examples/multiselect_native.rs"
|
|
with zipfile.ZipFile(archive_path) as archive:
|
|
source = archive.read(source_path)
|
|
need(pin(source) == parsed(inputs_data)[source_path], "historical checker source pin")
|
|
member = "checker-source/" + pin(source)["sha256"] + ".rs"
|
|
add(member, source)
|
|
spec["checkerSource"] = {"inputPath": source_path, "member": member}
|
|
specs[run] = spec
|
|
manifest = {"schema": 1, "kind": "multiselect-native", "runs": specs,
|
|
"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 stream:
|
|
stream.write((json.dumps(value, indent=2) + "\n").encode())
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", type=Path, default=HERE.parents[3])
|
|
parser.add_argument("--runs", nargs="+", required=True)
|
|
parser.add_argument("--output", type=Path, default=HERE)
|
|
parser.add_argument("--check-only", action="store_true")
|
|
args = parser.parse_args()
|
|
need(len(args.runs) == len(set(args.runs)) and all(re.fullmatch(r"run-[0-9]+", name) for name in args.runs), "explicit unique run-N names required")
|
|
payload, manifest, summary = collect(args.root.resolve(), args.runs)
|
|
if args.check_only:
|
|
print("PASS raw consistency; no archive 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 output refused: " + name)
|
|
for name in ("prepare.py", "verify.py", "README.md"):
|
|
data = (HERE / name).read_bytes()
|
|
if (out / name).exists():
|
|
need((out / name).read_bytes() == data, "different existing package tool")
|
|
else:
|
|
with (out / name).open("xb") as stream:
|
|
stream.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 evidence")
|
|
print("PASS package readback:", out, summary)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|