"""Verify archived native receipts and bytes; do not rerun the GUI or checker.""" from pathlib import Path, PurePosixPath from datetime import datetime import argparse import hashlib import json import io import re import stat import zipfile BASE = {"prepared.json", "started.json", "finished.json", "generation.json", "generation.stdout.log", "generation.stderr.log", "app.stdout.log", "app.stderr.log", "capture-executed.js"} OUTER = {"prepare.py", "verify.py", "README.md", "manifest.json", "evidence.zip"} def need(ok, message): if not ok: raise ValueError(message) def pin(data): return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()} def safe(name): return (isinstance(name, str) and bool(name) and "\\" not in name and ":" not in name and not PurePosixPath(name).is_absolute() and all(ord(c) >= 32 for c in name) and all(part not in ("", ".", "..") for part in name.split("/"))) def parsed(data): def unique(items): result = {} for key, value in items: need(key not in result, "duplicate JSON key: " + key) result[key] = value return result return json.loads(data, object_pairs_hook=unique) def norm(path): return path.replace("\\", "/").rstrip("/") def relative(path, root): prefix = norm(root) + "/" need(norm(path).startswith(prefix), "recorded path outside owner: " + path) value = norm(path)[len(prefix):] need(safe(value), "unsafe recorded relative path") return value def ordered(*values): times = [datetime.fromisoformat(value) for value in values] return all(t.utcoffset() is not None for t in times) and times == sorted(times) def zip_inventory(archive, members): names = archive.namelist() need(len(names) == len(set(names)) and set(names) == set(members), "exact archive members") for info in archive.infolist(): need(safe(info.filename) and not info.is_dir() and not stat.S_ISLNK(info.external_attr >> 16), "unsafe archive member") need(info.file_size == members[info.filename]["bytes"], "archive member size") def verify_data(manifest, read): need(manifest["schema"] == 1 and manifest["kind"] == "process-multiselect-native", "manifest kind") members, runs = manifest["members"], manifest["runs"] need(bool(runs) and all(re.fullmatch(r"run-[0-9]+", run) for run in runs), "run names") need(all(safe(name) for name in members), "safe member names") for name, expected in members.items(): need(pin(read(name)) == expected, "member pin: " + name) used, summaries = set(), {} def raw(name): need(name in members, "missing member: " + name) used.add(name) return read(name) def obj(name): return parsed(raw(name)) def tree(prefix): return {name[len(prefix):]: raw(name) for name in members if name.startswith(prefix)} def pinned_tree(prefix, expected): actual = tree(prefix) need({name: pin(data) for name, data in actual.items()} == expected, "pinned tree: " + prefix) return actual for run, spec in runs.items(): prefix = "runs/" + run + "/raw/" prepared, start, end, generation = [obj(prefix + n + ".json") for n in ("prepared", "started", "finished", "generation")] need(spec["reviewStatus"] in ("failed", "unreviewed") and isinstance(spec["reviewNote"], str) and spec["reviewNote"], "explicit non-acceptance status") if run == "run-1": need(spec["reviewStatus"] == "failed", "run-1 must retain observed native failure") origin = norm(prepared["cwd"]) + "/.local/process-multiselect-native/" + run need(spec["originalRun"] == origin and prepared["run"] == run and start["cwd"] == prepared["cwd"], "original run mapping") need(start["prepared"] == pin(raw(prefix + "prepared.json")), "launch preparation pin") need(ordered(generation["start"], generation["finish"], prepared["prepared"], start["start"], end["finish"]), "native lifetime") need(type(start["pid"]) is int and start["pid"] > 0 and type(end["exitCode"]) is int, "closed native process") for key in ("guiUnchanged", "checkerUnchanged", "fixtureUnchanged"): need(type(end[key]) is bool, "closure flag") need(start["gui"] == prepared["gui"] and start["checker"] == prepared["checker"] == generation["checker"], "binary identity") need(start["command"] == [prepared["gui"]["binary"]], "GUI command") need(generation["exitCode"] == 0 and generation["command"][:2] == [prepared["checker"]["binary"], "generate"] and len(generation["command"]) == 3 and norm(generation["command"][2]) == origin + "/fixture", "generation command") for stream in ("stdout", "stderr"): need(end[stream] == pin(raw(prefix + "app." + stream + ".log")), "GUI log pin") need(generation[stream] == pin(raw(prefix + "generation." + stream + ".log")), "generation log pin") need(set(spec["helpers"]) == set(prepared["helpers"]), "exact prepared helper mapping") for name, expected in prepared["helpers"].items(): need(safe(name) and "/" not in name and pin(raw(spec["helpers"][name])) == expected, "historical helper: " + name) need(start["helper"] == prepared["helpers"]["native.py"], "launch helper") need(pin(raw(prefix + "capture-executed.js")) == spec["executedCapture"], "actual capture function pin") fixture = pinned_tree(prefix + "fixture/", prepared["fixture"]) fm = parsed(fixture["manifest.json"]) need(fm["fixture"] == "process-multiselect-native-20260923" and fm["source_entry"] == prepared["entry"] and safe(prepared["entry"]), "fixture identity/entry") need(parsed(fixture["baseline.json"])["id"] == fm["ids"]["project"], "complete Project baseline identity") fixture_source = {name[7:]: data for name, data in fixture.items() if name.startswith("source/")} need(prepared["entry"] in fixture_source and norm(prepared["workingEntry"]) == origin + "/working/" + prepared["entry"], "working entry mapping") working = tree(prefix + "working/") need(bool(working) and all(name.endswith(".arcadia") for name in working), "saved working tree") build_maps = {} for kind, target in (("gui", "arcadia-studio"), ("checker", "process_multiselect_native")): recorded, b = prepared[kind], spec["builds"][kind] result, began, process = [obj(b[n]) for n in ("result.json", "started.json", "process.json")] need(pin(raw(b["result.json"])) == recorded["receiptPin"] and relative(recorded["receipt"], prepared["cwd"]) == b["originalRelative"], "build receipt mapping") need(all(result.get(k) == v for k, v in began.items()) and process["command"] == result["command"] and type(process["pid"]) is int and process["pid"] > 0, "build start/result/process") need(result["exitCode"] == 0 and result["inputsUnchanged"] is True and ordered(result["start"], process["observed"], result["finish"], generation["start"]), "successful preceding build") for stream in ("stdout", "stderr"): need(pin(raw(b[stream + ".log"])) == result[stream], "build log pin") need(pin(raw(b["helper"])) == result["helper"], "build helper pin") need(pin(raw(b["inputs.json"])) == recorded["inputs"] == result["inputs"], "build source map pin") inputs = obj(b["inputs.json"]) need(len(inputs) == result["inputCount"] and all(safe(name) for name in inputs), "source map count/paths") capsule = raw(b["inputs.zip"]) need(pin(capsule) == result["archive"], "build source capsule pin") with zipfile.ZipFile(io.BytesIO(capsule)) as archive: zip_inventory(archive, inputs) need(all(pin(archive.read(name)) == expected for name, expected in inputs.items()), "all captured source bytes") if kind == "checker": path = "crates/oa-core/examples/process_multiselect_native.rs" need(raw(spec["checkerSource"]) == archive.read(path), "historical checker source") artifacts = [row for row in result["artifacts"] if row["cargo"]["target"]["name"] == target] need(len(artifacts) == 1 and artifacts[0]["readback"] == recorded["pin"] and artifacts[0]["cargo"]["executable"] == recorded["binary"] and artifacts[0]["cargo"]["profile"]["test"] is False, "native executable provenance") stdout_rows = [parsed(line) for line in raw(b["stdout.log"]).splitlines() if line.strip().startswith(b"{")] need(artifacts[0]["cargo"] in stdout_rows, "artifact appears in raw Cargo output") build_maps[kind] = inputs need(prepared["gui"]["inputs"] == prepared["checker"]["inputs"] and build_maps["gui"] == build_maps["checker"], "same GUI/checker source map") top = {name[len(prefix):] for name in members if name.startswith(prefix) and "/" not in name[len(prefix):]} base = BASE | (top & {"history-bytes.json", "observations.md"}) if "observations.md" in top: raw(prefix + "observations.md") captures = sorted(name for name in top - base if name.endswith(".json")) need(captures, "native captures required") expected_top = base | set(captures) for name in captures: capture = obj(prefix + name) need(ordered(start["start"], capture["time"], end["finish"]), "capture lifetime") window = capture["window"] need(norm(window["app"]) == "process:" + norm(prepared["gui"]["binary"]) and type(window["id"]) is int and window["id"] > 0, "captured executable/window") need(isinstance(capture["accessibility"], dict) and isinstance(capture["images"], list), "accessibility/capture payload") for image in capture["images"]: filename = image["file"] need(safe(filename) and "/" not in filename and filename.startswith(name[:-5] + "-") and filename.endswith((".jpg", ".png")) and filename not in expected_top, "unique image mapping") need(all(type(image[k]) is int and image[k] > 0 for k in ("width", "height")), "image dimensions") expected_top.add(filename) data = raw(prefix + filename) need(data.startswith(b"\xff\xd8") if filename.endswith(".jpg") else data.startswith(b"\x89PNG\r\n\x1a\n"), "image signature") need(top == expected_top, "exact top-level run files") labels = {name[len(prefix + "snapshots/"):].split("/")[0] for name in members if name.startswith(prefix + "snapshots/")} states, results, positive = {}, {}, [] for label in sorted(labels): need(re.fullmatch(r"[a-z0-9-]+", label), "snapshot label") folder = prefix + "snapshots/" + label + "/" began, result = obj(folder + "started.json"), obj(folder + "result.json") need(all(result.get(k) == v for k, v in began.items()), "snapshot start/result") need(result["prepared"] == start["prepared"] and result["checker"] == prepared["checker"] and result["helper"] == prepared["helpers"]["snapshot.py"], "snapshot provenance") need(ordered(start["start"], result["start"], result["finish"], end["finish"]), "snapshot lifetime") mode, command = result["mode"], result["command"] need(mode in ("unchanged", "reordered") and len(command) == 5 and command[:3] == [prepared["checker"]["binary"], "check", mode] and norm(command[3]) == origin + "/fixture/baseline.json" and norm(command[4]) == origin + "/snapshots/" + label + "/source/" + prepared["entry"], "snapshot command") need(result["baseline"] == pin(fixture["baseline.json"]), "snapshot baseline") for stream in ("stdout", "stderr"): need(result[stream] == pin(raw(folder + stream + ".log")), "snapshot log pin") need(type(result["exitCode"]) is int and all(type(result[k]) is bool for k in ("binaryUnchanged", "workingSourceUnchanged", "snapshotUnchanged", "baselineUnchanged")), "snapshot closure") states[label] = pinned_tree(folder + "source/", result["source"]) need({name for name in members if name.startswith(folder)} == {folder + n for n in ("started.json", "result.json", "stdout.log", "stderr.log")} | {folder + "source/" + n for n in states[label]}, "exact snapshot files") if result["exitCode"] == 0: report = obj(folder + "stdout.log") need(report["result"] == "PASS" and report["comparison"] == "complete Project equality; literal expected order" and report["mode"] == mode and report["baseline"] == command[3] and report["saved"] == command[4], "positive complete Project report") positive.append(label) results[label] = result history_present = "history-bytes.json" in top if history_present: history = obj(prefix + "history-bytes.json") groups = history["equalByteGroups"] need(len(groups) == 2 and groups[0][:2] == ["baseline", "undo"] and groups[1] == ["reordered", "redo", "reopened"], "literal Process history groups") history_labels = [label for group in groups for label in group] need(len(history_labels) == len(set(history_labels)) and set(history_labels) == set(history["snapshots"]) and set(history_labels) <= set(positive), "history labels/positive reports") need(history["status"] == "PASS" and history["prepared"] == start["prepared"] and history["helper"] == prepared["helpers"]["history-bytes.py"] and ordered(end["finish"], history["verified"]), "history provenance") need(end["exitCode"] == 0 and all(end[k] is True for k in ("guiUnchanged", "checkerUnchanged", "fixtureUnchanged")), "history clean closure") for group, mode in zip(groups, ("unchanged", "reordered")): for label in group: need(states[label] == states[group[0]] and results[label]["mode"] == mode and history["snapshots"][label] == pin(raw(prefix + "snapshots/" + label + "/result.json")), "literal history bytes/receipt") need(all(results[label][k] is True for k in ("binaryUnchanged", "workingSourceUnchanged", "snapshotUnchanged", "baselineUnchanged")), "stable history snapshot") need(len(states[label]) == history["sourceFilesPerSnapshot"], "history source count") need(states["baseline"] == fixture_source and states["baseline"] != states["reordered"], "baseline exact and reorder changed") changed = sorted(name for name in set(states["baseline"]) | set(states["reordered"]) if states["baseline"].get(name) != states["reordered"].get(name)) need(changed == history["changedFilesFromBaseline"] == ["oa/behavior/process_oa_process.arcadia"], "only intended OA source changed") sequence = [results[label] for label in ("baseline", "reordered", "undo", "redo", "reopened")] need(all(ordered(a["finish"], b["start"]) for a, b in zip(sequence, sequence[1:])), "history chronology") summaries[run] = {"reviewStatus": spec["reviewStatus"], "exitCode": end["exitCode"], "captures": len(captures), "snapshots": len(labels), "positiveReports": len(positive), "historyBytes": history_present} need(used == set(members), "unreferenced package members: " + repr(set(members) - used)) return summaries def verify_package(root): files = parsed((root / "files.json").read_bytes()) need(set(files) == OUTER, "outer files") for name, expected in files.items(): need(pin((root / name).read_bytes()) == expected, "outer pin: " + name) manifest = parsed((root / "manifest.json").read_bytes()) with zipfile.ZipFile(root / "evidence.zip") as archive: zip_inventory(archive, manifest["members"]) return verify_data(manifest, archive.read) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("package", nargs="?", type=Path, default=Path(__file__).resolve().parent) args = parser.parse_args() print("PASS byte/receipt consistency only:", verify_package(args.package)) print("No GUI/checker execution or native acceptance claim; see independent parent review.")