VoxelForge/Standalone/Research/WorldAssemblerM3/Linux74/FalsifierL74.py

#!/usr/bin/env python3
"""Compare full-grid and proof-skip mesh hashes for all four owner arrivals."""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import time
from pathlib import Path


ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l74/logs"
DATA = ROOT / "Standalone/data/owner_world.json"
BINS = {
    "clang": ROOT / "Saved/Linux/l74/bin/VoxelForgeCoreHeadless-l74-clang",
    "gcc": ROOT / "Saved/Linux/l74/bin-gcc/VoxelForgeCoreHeadless-l74-gcc",
}
CAMERAS = {
    0: (8.0, 16.0, -372.2626953125),
    1: (25.8369140625, 3.19921875, -737.0),
    2: (19.166015625, -14.2001953125, -753.3740234375),
    3: (-20.0625, -5.068359375, -900.353515625),
}
HASH_ROW = re.compile(
    r"\[PipelineTileHash\] lod=(\d+) x=(-?\d+) y=(-?\d+) z=(-?\d+) "
    r"density=(0x[0-9a-fA-F]+) mesh=(0x[0-9a-fA-F]+) "
    r"payload=(0x[0-9a-fA-F]+) topology=(0x[0-9a-fA-F]+)"
)
CEILING_ROW = re.compile(
    r"\[UniformGridCeiling\] strate=(-?\d+) lod=(\d+) key=\((-?\d+),(-?\d+),(-?\d+)\)"
)


def read_hashes(path: Path) -> tuple[dict, set[tuple[int, int, int, int]]]:
    hashes = {}
    generated_density = set()
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        row = HASH_ROW.search(line)
        if row:
            lod, x, y, z = (int(row.group(i)) for i in range(1, 5))
            hashes[(lod, x, y, z)] = {
                "density": row.group(5).lower(), "mesh": row.group(6).lower(),
                "payload": row.group(7).lower(), "topology": row.group(8).lower(),
            }
        ceiling = CEILING_ROW.search(line)
        if ceiling:
            generated_density.add(tuple(int(ceiling.group(i)) for i in range(2, 6)))
    return hashes, generated_density


def run(compiler: str, view: int, workers: int, allow_skips: bool,
        negative_control: bool = False) -> Path:
    camera = ",".join(f"{value:.12f}" for value in CAMERAS[view]) + ",0,0"
    mode = "negative" if negative_control else "skip" if allow_skips else "reference"
    log = OUT / f"{mode}-{compiler}-view{view}-w{workers}.log"
    command = [
        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "timeout", "1740s", str(BINS[compiler]), str(DATA),
        "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
        "--force-generate", f"--workers={workers}", "--clip-radius=3",
        "--max-clip-level=8", "--max-level=8", f"--camera={camera}",
        "--stream-hash-tiles-allow-skips" if allow_skips or negative_control
        else "--stream-hash-tiles",
    ]
    import os
    environment = os.environ.copy()
    if negative_control:
        environment["VOXELFORGE_L74_TIGHT_PROOF_CONTROL"] = "1"
    started = time.monotonic()
    with log.open("w", encoding="utf-8") as output:
        result = subprocess.run(command, cwd=ROOT, env=environment, stdout=output,
                                stderr=subprocess.STDOUT, check=False, timeout=1740)
    hashes, _ = read_hashes(log)
    print(json.dumps({"compiler": compiler, "view": view, "workers": workers,
                      "mode": mode, "exit_code": result.returncode,
                      "hash_rows": len(hashes), "wall_s": round(time.monotonic()-started, 3),
                      "log": str(log.relative_to(ROOT))}, sort_keys=True), flush=True)
    if result.returncode != 0 or not hashes:
        raise SystemExit(f"hash generation failed; see {log}")
    return log


def compare(reference: Path, candidate: Path) -> dict:
    reference_hashes, _ = read_hashes(reference)
    candidate_hashes, candidate_generated = read_hashes(candidate)
    missing = sorted(reference_hashes.keys() - candidate_hashes.keys())
    extra = sorted(candidate_hashes.keys() - reference_hashes.keys())
    mismatch_counts = {field: 0 for field in ("mesh", "payload", "topology", "density_processed")}
    mismatch_examples = []
    for key in sorted(reference_hashes.keys() & candidate_hashes.keys()):
        a, b = reference_hashes[key], candidate_hashes[key]
        fields = ["mesh", "payload", "topology"]
        if key in candidate_generated:
            fields.append("density")
        for field in fields:
            if a[field] != b[field]:
                label = "density_processed" if field == "density" else field
                mismatch_counts[label] += 1
                if len(mismatch_examples) < 8:
                    mismatch_examples.append({"key": key, "field": field,
                                              "reference": a[field], "candidate": b[field]})
    return {"reference": str(reference.relative_to(ROOT)),
            "candidate": str(candidate.relative_to(ROOT)),
            "reference_tiles": len(reference_hashes), "candidate_tiles": len(candidate_hashes),
            "missing_keys": missing[:8], "missing_key_count": len(missing),
            "extra_keys": extra[:8], "extra_key_count": len(extra),
            "mismatch_counts": mismatch_counts, "mismatch_examples": mismatch_examples,
            "mesh_topology_equal": not missing and not extra
                and mismatch_counts["mesh"] == 0 and mismatch_counts["topology"] == 0}


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--phase", choices=("matrix", "workers", "negative"), required=True)
    args = parser.parse_args()
    OUT.mkdir(parents=True, exist_ok=True)
    missing = [str(path) for path in [DATA, *BINS.values()] if not path.is_file()]
    if missing:
        raise SystemExit("missing input: " + ", ".join(missing))
    comparisons = []
    if args.phase == "matrix":
        references = {}
        skip_eight = {}
        for compiler in ("clang", "gcc"):
            for view in range(4):
                reference = run(compiler, view, 8, False)
                skip = run(compiler, view, 8, True)
                references[(compiler, view)] = reference
                skip_eight[(compiler, view)] = skip
                comparisons.append(compare(reference, skip))
        for view in range(4):
            comparisons.append(compare(references[("clang", view)],
                                      references[("gcc", view)]))
            comparisons.append(compare(skip_eight[("clang", view)],
                                      skip_eight[("gcc", view)]))
        (OUT / "falsifier-matrix.json").write_text(json.dumps(comparisons, indent=2)+"\n",
                                                    encoding="utf-8")
    elif args.phase == "workers":
        for compiler in ("clang", "gcc"):
            for view in range(4):
                reference = OUT / f"reference-{compiler}-view{view}-w8.log"
                if not reference.is_file():
                    raise SystemExit(f"missing reference from matrix phase: {reference}")
                baseline = OUT / f"skip-{compiler}-view{view}-w8.log"
                for workers in (1, 12):
                    candidate = run(compiler, view, workers, True)
                    comparisons.append(compare(reference, candidate))
                    comparisons.append(compare(baseline, candidate))
        (OUT / "falsifier-workers.json").write_text(json.dumps(comparisons, indent=2)+"\n",
                                                     encoding="utf-8")
    else:
        for view in range(4):
            reference = OUT / f"reference-clang-view{view}-w8.log"
            candidate = run("clang", view, 8, True, negative_control=True)
            comparisons.append(compare(reference, candidate))
        (OUT / "falsifier-negative-control.json").write_text(
            json.dumps(comparisons, indent=2)+"\n", encoding="utf-8")
        caught = any(not row["mesh_topology_equal"] for row in comparisons)
        print(f"[L74NegativeControlResult] caught={int(caught)}", flush=True)
        if not caught:
            return 2
        return 0
    failures = [row for row in comparisons if not row["mesh_topology_equal"]]
    print(f"[L74Falsifier] phase={args.phase} comparisons={len(comparisons)} "
          f"mesh_topology_failures={len(failures)}", flush=True)
    return 0 if not failures else 1


if __name__ == "__main__":
    raise SystemExit(main())