283 lines
15 KiB
Python
283 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Review a proposed HD family with the exact already-built Rust loader.
|
|
|
|
Creates an isolated temporary pack with hard-linked raw artwork. It never edits
|
|
the active manifest or generated pixels. Build the game library before use.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from PIL import Image, ImageDraw
|
|
|
|
from hd_frame_policy import FRAME_KEYS, validate_frame_policy
|
|
|
|
|
|
def proposal_frames(proposal, expected=None):
|
|
"""Check source-bound target membership; artistic approval remains separate."""
|
|
rows = {}
|
|
for sequence in proposal.get("sequences", []):
|
|
base = sequence["base_asset"]
|
|
if isinstance(base, str):
|
|
base = int(base, 16)
|
|
if not 1 <= sequence["frame_count"] <= 512:
|
|
raise ValueError("invalid sequence frame count")
|
|
for frame in sequence["frames"]:
|
|
if set(frame) - FRAME_KEYS:
|
|
raise ValueError("unknown manifest frame fields")
|
|
validate_frame_policy(frame)
|
|
if not 0 <= frame["frame"] < sequence["frame_count"]:
|
|
raise ValueError("frame outside sequence")
|
|
identity = f"{base + frame['frame']:05x}"
|
|
if identity in rows:
|
|
raise ValueError("duplicate candidate ID")
|
|
rows[identity] = frame
|
|
if not rows:
|
|
raise ValueError("candidate needs manifest-compatible sequences")
|
|
supplied = {str(row["asset"]).lower().removeprefix("0x").zfill(5): row for row in proposal.get("targets", [])}
|
|
if supplied and set(supplied) != set(rows):
|
|
raise ValueError("candidate targets and sequences differ")
|
|
for identity, frame in rows.items():
|
|
if supplied and any(supplied[identity].get(key) != value for key, value in frame.items()):
|
|
raise ValueError(f"target and sequence metadata differ for {identity}")
|
|
if expected is not None:
|
|
if identity not in expected:
|
|
raise ValueError(f"candidate claims foreign ID {identity}")
|
|
source = expected[identity]
|
|
if frame["source_sha256"] != source["source_sha256"]:
|
|
raise ValueError(f"candidate source changed for {identity}")
|
|
if [frame["logical_width"], frame["logical_height"]] != source["logical_canvas"]:
|
|
raise ValueError(f"candidate logical canvas changed for {identity}")
|
|
if frame["origin"] != [0, 0]:
|
|
raise ValueError(f"candidate anchor changed for {identity}")
|
|
relative = Path(frame["file"])
|
|
if relative.is_absolute() or ".." in relative.parts:
|
|
raise ValueError("artwork must remain pack-relative")
|
|
if expected is not None and proposal.get("complete_family") and set(rows) != set(expected):
|
|
raise ValueError("complete-family claim omits source IDs")
|
|
return rows
|
|
|
|
|
|
def binding_digest(rows):
|
|
"""Hash only renderer-bound fields, separately from artist review notes."""
|
|
return hashlib.sha256(json.dumps(rows, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def freeze_library(project, library):
|
|
"""Keep exact tested bytes and the original build time across Cargo rebuilds."""
|
|
before = library.stat()
|
|
payload = library.read_bytes()
|
|
after = library.stat()
|
|
identity = lambda value: (value.st_ino, value.st_size, value.st_mtime_ns)
|
|
if identity(before) != identity(after):
|
|
raise ValueError("Rust library changed while taking review snapshot; retry")
|
|
digest = hashlib.sha256(payload).hexdigest()
|
|
directory = project / "target/hd-review-libraries" / digest
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
frozen = directory / library.name
|
|
if frozen.exists():
|
|
if hashlib.sha256(frozen.read_bytes()).hexdigest() != digest:
|
|
raise ValueError("review library snapshot is corrupt")
|
|
if frozen.stat().st_mtime_ns != before.st_mtime_ns:
|
|
# Identical bytes rebuilt later must not make an old snapshot look
|
|
# fresh. Separate build times retain an honest source-age check.
|
|
frozen = directory / (library.stem + "-build-" + str(before.st_mtime_ns) + library.suffix)
|
|
if not frozen.exists():
|
|
descriptor, temporary = tempfile.mkstemp(prefix="library-", dir=directory)
|
|
try:
|
|
with os.fdopen(descriptor, "wb") as output:
|
|
output.write(payload)
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
os.utime(temporary, ns=(before.st_atime_ns, before.st_mtime_ns))
|
|
os.chmod(temporary, 0o444)
|
|
try:
|
|
# Publish without replacing an identical snapshot another
|
|
# reviewer may already be using. Concurrent first readers
|
|
# must retain one stable inode as well as identical bytes.
|
|
os.link(temporary, frozen)
|
|
except FileExistsError:
|
|
pass
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
if hashlib.sha256(frozen.read_bytes()).hexdigest() != digest or frozen.stat().st_mtime_ns != before.st_mtime_ns:
|
|
raise ValueError("review library snapshot identity differs")
|
|
return frozen, digest, before
|
|
|
|
|
|
def select_compatible_serde_json(project, library, dependency_directory, output, candidates=None):
|
|
"""Compile-check serde's trait identity against the exact frozen game library.
|
|
|
|
Cargo can leave several feature graphs in one deps directory. A newer
|
|
serde_json may use a different serde_core instance from Manifest, so its
|
|
build time cannot establish compatibility.
|
|
"""
|
|
candidates = sorted(candidates if candidates is not None else
|
|
dependency_directory.glob("libserde_json-*.rlib"))
|
|
if not candidates:
|
|
raise ValueError("build serde_json before candidate review")
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
report = {
|
|
"method": "rustc metadata compile of Manifest Deserialize and Serialize against exact frozen libraries",
|
|
"raptor_library_sha256": hashlib.sha256(library.read_bytes()).hexdigest(),
|
|
"attempts": [],
|
|
}
|
|
evidence = output / "dependency-selection.json"
|
|
with tempfile.TemporaryDirectory(prefix="serde-compatibility-", dir=output) as temporary:
|
|
directory = Path(temporary)
|
|
source = directory / "probe.rs"
|
|
source.write_text("""fn main() {
|
|
let manifest: raptor_replica::hd::Manifest = serde_json::from_slice(b"{}").unwrap();
|
|
let _ = serde_json::to_vec(&manifest).unwrap();
|
|
}
|
|
""")
|
|
for candidate in candidates:
|
|
frozen, digest, original = freeze_library(project, candidate)
|
|
result = subprocess.run([
|
|
"rustc", "--edition=2021", "--emit=metadata", str(source),
|
|
"--extern", f"raptor_replica={library}",
|
|
"--extern", f"serde_json={frozen}",
|
|
"-L", f"dependency={dependency_directory}",
|
|
"-o", str(directory / "probe.rmeta"),
|
|
], capture_output=True, text=True)
|
|
report["attempts"].append({
|
|
"build_origin": str(candidate.relative_to(project)),
|
|
"library": str(frozen.relative_to(project)),
|
|
"sha256": digest,
|
|
"build_origin_modified_ns": original.st_mtime_ns,
|
|
"compatible": result.returncode == 0,
|
|
"compiler_stderr": result.stderr,
|
|
})
|
|
if result.returncode == 0:
|
|
report["selected"] = report["attempts"][-1]
|
|
evidence.write_text(json.dumps(report, indent=2) + "\n")
|
|
return frozen, report
|
|
evidence.write_text(json.dumps(report, indent=2) + "\n")
|
|
raise ValueError(f"no compiler-compatible serde_json for the selected Rust library; see {evidence}")
|
|
|
|
|
|
def main():
|
|
project = Path(__file__).resolve().parents[1]
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--candidate", type=Path, required=True)
|
|
parser.add_argument("--data", type=Path, default=project.parent / "data")
|
|
parser.add_argument("--output", type=Path)
|
|
parser.add_argument("--profile", choices=["auto", "debug", "release"], default="auto")
|
|
parser.add_argument("--sheet-rows", type=int, default=16, help="Maximum frames per contact-sheet page; review every page")
|
|
args = parser.parse_args()
|
|
if not 1 <= args.sheet_rows <= 64:
|
|
parser.error("sheet-rows must be 1..64")
|
|
candidate_at_start = args.candidate.read_bytes()
|
|
proposal = json.loads(candidate_at_start)
|
|
job_path = args.candidate.parent / "job.json"
|
|
expected = None
|
|
if job_path.is_file():
|
|
job = json.loads(job_path.read_text())
|
|
if proposal.get("job_id") != job["job_id"]:
|
|
raise ValueError("candidate job identity differs")
|
|
expected = {target["asset"]: target for target in job["targets"]}
|
|
rows = proposal_frames(proposal, expected)
|
|
bindings_at_start = binding_digest(rows)
|
|
static_cards = any(int(identity, 16) in [0x1002b, 0x1002d, 0x1002f] for identity in rows)
|
|
manifest = {"format_version": 1, "name": "Isolated candidate review", "description": "Unregistered diagnostic pack", "sequences": proposal["sequences"]}
|
|
profiles = ["debug", "release"] if args.profile == "auto" else [args.profile]
|
|
libraries = [library for profile in profiles for library in (project / "target" / profile / "deps").glob("libraptor_replica-*.rlib")]
|
|
if not libraries:
|
|
raise ValueError("build the Rust game library before candidate review")
|
|
built_library = max(libraries, key=lambda path: path.stat().st_mtime_ns)
|
|
dependency_directory = built_library.parent
|
|
library, library_sha256, built_library_at_start = freeze_library(project, built_library)
|
|
library_at_start = library.stat()
|
|
output = (args.output or args.candidate.parent / "runtime-review").resolve()
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
serde, serde_selection = select_compatible_serde_json(project, library, dependency_directory, output)
|
|
pack_root = (args.data / "hd").resolve()
|
|
artwork_hashes = {}
|
|
with tempfile.TemporaryDirectory(prefix="isolated-pack-", dir=output) as temporary:
|
|
isolated = Path(temporary)
|
|
for frame in rows.values():
|
|
source = (pack_root / frame["file"]).resolve()
|
|
if not source.is_relative_to(pack_root):
|
|
raise ValueError("artwork escapes real pack root")
|
|
artwork_hashes[frame["file"]] = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
destination = isolated / frame["file"]
|
|
if destination.exists():
|
|
continue
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
os.link(source, destination)
|
|
except OSError:
|
|
shutil.copyfile(source, destination)
|
|
(isolated / "hd.json").write_text(json.dumps(manifest))
|
|
binary = isolated / "review-bin"
|
|
compiler_args = ["rustc", "--edition=2021", str(project / "tools/hd_candidate_review.rs"), "--extern", f"raptor_replica={library}", "--extern", f"serde_json={serde}", "-L", f"dependency={dependency_directory}", "-L", "native=" + str(Path.home() / ".local/lib"), "-o", str(binary)]
|
|
if static_cards:
|
|
compiler_args.extend(["--cfg", "static_card_api"])
|
|
subprocess.run(compiler_args, check=True)
|
|
environment = os.environ.copy()
|
|
environment["LD_LIBRARY_PATH"] = str(Path.home() / ".local/lib") + os.pathsep + environment.get("LD_LIBRARY_PATH", "")
|
|
subprocess.run([str(binary), str(args.data.resolve()), str(isolated), str(output)], check=True, env=environment)
|
|
for relative, digest in artwork_hashes.items():
|
|
if hashlib.sha256((pack_root / relative).read_bytes()).hexdigest() != digest:
|
|
raise ValueError("artwork changed during runtime review")
|
|
library_at_finish = library.stat()
|
|
if (library_at_start.st_ino, library_at_start.st_size, library_at_start.st_mtime_ns) != (library_at_finish.st_ino, library_at_finish.st_size, library_at_finish.st_mtime_ns):
|
|
raise ValueError("Rust library changed during runtime review; rebuild and rerun")
|
|
qa_path = output / "runtime-qa.json"
|
|
qa = json.loads(qa_path.read_text())
|
|
candidate_at_finish = args.candidate.read_bytes()
|
|
if binding_digest(proposal_frames(json.loads(candidate_at_finish), expected)) != bindings_at_start:
|
|
raise ValueError("candidate renderer bindings changed during runtime review")
|
|
qa["library"] = str(library.relative_to(project))
|
|
qa["library_modified_ns"] = library_at_start.st_mtime_ns
|
|
qa["library_sha256"] = library_sha256
|
|
qa["library_build_origin"] = str(built_library.relative_to(project))
|
|
qa["library_build_origin_modified_ns"] = built_library_at_start.st_mtime_ns
|
|
qa["serde_json_dependency"] = serde_selection
|
|
qa["candidate_sha256"] = hashlib.sha256(candidate_at_finish).hexdigest()
|
|
qa["candidate_sha256_at_start"] = hashlib.sha256(candidate_at_start).hexdigest()
|
|
qa["candidate_notes_changed_during_review"] = candidate_at_start != candidate_at_finish
|
|
qa["frame_bindings_sha256"] = bindings_at_start
|
|
qa["artwork_sha256"] = artwork_hashes
|
|
qa["static_card_palette_mode"] = static_cards
|
|
qa["source_newer_than_library"] = any((project / "src" / name).stat().st_mtime_ns > library_at_start.st_mtime_ns for name in ["assets.rs", "hd.rs"])
|
|
qa["any_game_source_newer_than_library"] = any(path.stat().st_mtime_ns > library_at_start.st_mtime_ns for path in (project / "src").glob("*.rs"))
|
|
qa_path.write_text(json.dumps(qa, indent=2) + "\n")
|
|
identities = sorted(rows)
|
|
pages = []
|
|
# These are tool-produced diagnostics, never input art. Remove stale pages
|
|
# so a former shorter proposal cannot be mistaken for this complete review.
|
|
for obsolete in output.glob("runtime-contact-sheet*.png"):
|
|
obsolete.unlink()
|
|
for start in range(0, len(identities), args.sheet_rows):
|
|
page = identities[start:start + args.sheet_rows]
|
|
cell_width = max(rows[identity]["logical_width"] for identity in page) * 4 + 24
|
|
row_height = max(rows[identity]["logical_height"] for identity in page) * 4 + 42
|
|
sheet = Image.new("RGB", (cell_width * 3, row_height * len(page)), (20, 28, 38))
|
|
draw = ImageDraw.Draw(sheet)
|
|
for row, identity in enumerate(page):
|
|
for column, (suffix, label) in enumerate([("original", "Original"), ("hd-1", "Rust 1x"), ("hd-4", "Rust 4x")]):
|
|
with Image.open(output / f"{identity}-{suffix}.ppm") as source:
|
|
sprite = source.convert("RGB")
|
|
if suffix != "hd-4":
|
|
sprite = sprite.resize((sprite.width * 4, sprite.height * 4), Image.Resampling.NEAREST)
|
|
x, y = column * cell_width + 12, row * row_height + 36
|
|
draw.text((x, row * row_height + 4), identity, fill=(225, 225, 230))
|
|
draw.text((x, row * row_height + 16), label, fill=(225, 225, 230))
|
|
sheet.paste(sprite, (x, y))
|
|
name = "runtime-contact-sheet.png" if len(identities) <= args.sheet_rows else f"runtime-contact-sheet-{len(pages) + 1:03}.png"
|
|
sheet.save(output / name)
|
|
pages.append({"file": name, "assets": page})
|
|
(output / "runtime-contact-sheets.json").write_text(json.dumps({"review_every_page": True, "pages": pages}, indent=2) + "\n")
|
|
print(json.dumps({"output": str(output), "frames": len(rows), "sheet_pages": len(pages), "stale_library": qa["source_newer_than_library"]}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|