Files
ARCADIA-Studio/harness/experiments/state-multiselect-20260923/source/verify-committed-inputs.py
T

122 lines
5.6 KiB
Python

"""Compare captured input maps to exact named Git trees; no builds or checkouts."""
from pathlib import Path, PurePosixPath
import argparse
import hashlib
import json
import os
import subprocess
import zipfile
HERE = Path(__file__).resolve().parent
COMPARISON = "Exact committed Git blobs equal every captured input; same complete filtered inventory"
EXPLICIT_FIXTURES = {
"harness/experiments/edge-attachments-20260922/native-fixture.arcadia",
"harness/experiments/textual-authoring-20260911/alarm-response.arcadia",
}
INCLUDE_FIXTURES = False
def need(ok, message):
if not ok:
raise ValueError(message)
def pin(data):
return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}
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 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(part not in ("", ".", "..") for part in name.split("/")))
def relevant(path):
return (INCLUDE_FIXTURES and path in EXPLICIT_FIXTURES) or path in ("Cargo.toml", "Cargo.lock", "rust-toolchain.toml") or path.startswith(
("crates/", "build/", ".cargo/", "scripts/", "docs/", "examples/", "harness/fixtures/", "harness/prototypes/"))
def git(repo, *arguments, data=None):
environment = dict(os.environ, GIT_NO_LAZY_FETCH="1")
return subprocess.run(["git", *arguments], cwd=repo, input=data, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, check=True, env=environment).stdout
def compare_commit(repo, commit, inputs):
need(git(repo, "rev-parse", "--verify", commit + "^{commit}").decode().strip() == commit, "named commit unavailable")
entries = git(repo, "ls-tree", "-r", "--full-tree", "-z", commit).split(b"\0")
blobs = {}
for entry in filter(None, entries):
header, path_bytes = entry.split(b"\t", 1)
path = path_bytes.decode("utf-8")
if relevant(path):
_mode, kind, oid = header.split()
need(kind == b"blob", "relevant non-blob tree entry: " + path)
need(path not in blobs, "duplicate tree path")
blobs[path] = oid
need(set(blobs) == set(inputs), "complete filtered Git inventory differs; missing="
+ repr(sorted(set(inputs) - set(blobs))) + "; additional=" + repr(sorted(set(blobs) - set(inputs))))
# Object IDs avoid path quoting and commit:path interpretation in batch input.
paths = sorted(blobs)
data = git(repo, "cat-file", "--batch", data=b"".join(blobs[path] + b"\n" for path in paths))
offset, actual = 0, {}
for path in paths:
end = data.index(b"\n", offset)
oid, kind, size_bytes = data[offset:end].split()
size = int(size_bytes)
need(oid == blobs[path] and kind == b"blob" and size >= 0, "Git batch object relationship")
content = data[end + 1:end + 1 + size]
need(len(content) == size and data[end + 1 + size:end + 2 + size] == b"\n", "Git batch framing")
actual[path] = pin(content)
offset = end + 2 + size
need(offset == len(data), "trailing Git batch bytes")
different = [path for path in paths if actual[path] != inputs[path]]
need(not different, "captured bytes differ from named commit: " + repr(different))
def main():
import datetime
global INCLUDE_FIXTURES
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repo", type=Path, required=True)
parser.add_argument("--raw-gates", type=Path, required=True)
parser.add_argument("--gate", required=True)
parser.add_argument("--commit", required=True, help="exact full commit object ID, not a branch")
parser.add_argument("--include-fixtures", action="store_true", help="include both compile-time fixtures for the revised 735-input scope; default preserves historical 733-input scope")
parser.add_argument("--output", type=Path, help="optional new receipt file; otherwise read-only check")
args = parser.parse_args()
INCLUDE_FIXTURES = args.include_fixtures
need(safe(args.gate) and "/" not in args.gate, "safe gate name")
folder = args.raw_gates / args.gate
map_bytes = (folder / "inputs.json").read_bytes()
result = parsed((folder / "result.json").read_bytes())
inputs = parsed(map_bytes)
need(result["exitCode"] == 0 and result["inputsUnchanged"] is True, "accepted unchanged gate required")
need(result["inputs"] == pin(map_bytes) and result["inputCount"] == len(inputs), "map relationship")
need(all(safe(path) and relevant(path) for path in inputs), "input scope")
need(sum(path.startswith("crates/vendor/iced_widget/") for path in inputs) == 71, "vendor inventory")
compare_commit(args.repo, args.commit, inputs)
report = {"checked": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"commit": args.commit, "gate": args.gate, "inputCount": len(inputs),
"inputMap": pin(map_bytes), "comparison": COMPARISON,
"explicitCompileTimeFixtures": sorted(EXPLICIT_FIXTURES) if INCLUDE_FIXTURES else []}
if args.output:
with args.output.open("x", encoding="utf-8") as stream:
json.dump(report, stream, indent=2)
print("PASS exact Git tree/blobs:", json.dumps(report))
print("Source identity only; separately verify capture/archive consistency. No build, checkout or native acceptance claim.")
if __name__ == "__main__":
main()