VoxelForge/Standalone/Research/WorldAssemblerM3/Linux176/RunL176SeedSweep.py

#!/usr/bin/env python3
"""Run L176 step-3 audits and route gates for owner seeds 0..15."""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l176/seeds"
BIN = ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-l176-clang"
DATA = ROOT / "Standalone/data/owner_world.json"
STAMP_TOOL = ROOT / "Standalone/Tools/verify/source_stamp.py"
STAMP_RE = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]+)$", re.M)


def one(pattern: str, content: str) -> re.Match[str] | None:
    return re.search(pattern, content, re.M)


def evaluate(content: str, expected_stamp: str, require_l130: bool) -> dict[str, object]:
    stamp = STAMP_RE.search(content)
    seam = one(r"^\[PassageTileSeamAudit\].*$", content)
    seam_fields = dict(re.findall(r"\b([a-z0-9_]+)=([^ ]+)", seam.group(0))) if seam else {}
    pad = one(r"^\[LandingPadAuditSummary\].*$", content)
    air = one(r"^\[NoAddedSolidOverHostAirAudit\].*$", content)
    floor = one(r"^\[PassageFloorShellAuditSummary\].*$", content)
    interior = one(r"^\[PassageInteriorCapsuleAuditSummary\].*$", content)
    end_run = one(r"^\[PassageEndRunAuditSummary\].*$", content)
    l130 = one(r"^\[L130PassageReachProofAudit\].*$", content)
    route_rows = {}
    for name, strate in (("tunnel_network", 2), ("vertical_shafts", 3)):
        route_rows[name] = one(rf"^\[ResolvedStrateRouteAudit\] strate={strate} .*status=(PASS|FAIL)$", content)
    controls = {
        name: bool(one(rf"^\[{label}BlockedRouteControl\].*expected=FAIL status=PASS$", content))
        for name, label in (("tunnel_network", "TunnelNetwork"),
                            ("vertical_shafts", "VerticalShafts"))
    }
    checks = {
        "source_stamp": bool(stamp and stamp.group(1) == expected_stamp),
        "seam_zero_mismatches": seam_fields.get("mismatches") == "0",
        "seam_no_exempt_pairs": seam_fields.get("exempt_pairs") == "0",
        "landing_pads": bool(pad and re.search(
            r"pads=6 passed=6 negative_controls=24/24 status=PASS$", pad.group(0))),
        "no_added_solid_over_host_air": bool(air and re.search(
            r"added_solid_samples=0 .*negative_control=DETECTED status=PASS$", air.group(0))),
        "floor_shell": bool(floor and "status=PASS" in floor.group(0)),
        "passage_interior_air": bool(interior and "status=PASS" in interior.group(0)),
        "end_runs": bool(end_run and re.search(
            r"passages=3 measured=3 passed=3 status=PASS$", end_run.group(0))),
        "tunnel_network_route": bool(route_rows["tunnel_network"] and
            "status=PASS" in route_rows["tunnel_network"].group(0)),
        "vertical_shafts_route": bool(route_rows["vertical_shafts"] and
            "status=PASS" in route_rows["vertical_shafts"].group(0)),
        "route_blocked_controls": all(controls.values()),
    }
    if l130:
        fields = dict(re.findall(r"\b([a-z0-9_]+)=([^ ]+)", l130.group(0)))
        checks["l130_pad_bounds"] = (fields.get("status") == "PASS"
            and fields.get("pad_proof_misses") == "0"
            and fields.get("pad_feature_misses") == "0"
            and fields.get("pad_stamp_misses") == "0")
    elif require_l130:
        checks["l130_pad_bounds"] = False
    else:
        checks["l130_pad_bounds"] = True
    return {
        "checks": checks,
        "step3_passed": all(checks.values()),
        "seam_summary": seam_fields,
        "route_rows": {key: value.group(0) if value else None
                        for key, value in route_rows.items()},
        "resolved_passage_summary": (one(r"^\[ResolvedPassageAuditSummary\].*$", content).group(0)
                                      if one(r"^\[ResolvedPassageAuditSummary\].*$", content) else None),
        "l130_line": l130.group(0) if l130 else None,
    }


def main() -> int:
    import argparse
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--seed-start", type=int, default=0)
    parser.add_argument("--seed-stop", type=int, default=16, help="exclusive")
    parser.add_argument("--no-resume", action="store_true")
    args = parser.parse_args()
    if not 0 <= args.seed_start <= args.seed_stop <= 16:
        raise SystemExit("seed range must be within 0..16")
    if not BIN.is_file() or not DATA.is_file():
        raise SystemExit(f"missing binary or owner fixture: {BIN}")
    OUT.mkdir(parents=True, exist_ok=True)
    stamp = subprocess.check_output(
        [sys.executable, str(STAMP_TOOL), "stamp", "--root", str(ROOT)], text=True).strip()
    rows: list[dict[str, object]] = []
    for seed in range(args.seed_start, args.seed_stop):
        log, layout, exit_file = (OUT / f"seed{seed}.log", OUT / f"seed{seed}.layout.json",
                                 OUT / f"seed{seed}.exit")
        command = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
            "timeout", "1740s", str(BIN), str(DATA), "--world-assembler-prototype",
            "--stream-benchmark", "--stream-pipeline", "--workers=8", "--clip-radius=3",
            "--max-clip-level=8", "--max-level=0", "--force-generate", f"--seed={seed}",
            "--recipe=surface-world", "--density-dispatch-report",
            "--audit-resolved-layout-passages", "--audit-passage-tile-seams",
            f"--resolved-layout-output={layout}"]
        env = os.environ.copy()
        if seed == 0:
            env["VOXELFORGE_L130_PASSAGE_REACH_PROOF_AUDIT"] = "1"
        resumed = False
        started = time.monotonic()
        if not args.no_resume and log.is_file() and layout.is_file() and exit_file.is_file():
            code: int | str = int(exit_file.read_text().strip())
            resumed = True
        else:
            try:
                with log.open("w", encoding="utf-8") as output:
                    proc = subprocess.run(command, cwd=ROOT, env=env, stdout=output,
                        stderr=subprocess.STDOUT, check=False, timeout=1740)
                code = proc.returncode
            except subprocess.TimeoutExpired:
                code = "OUTER_TIMEOUT"
            if isinstance(code, int):
                exit_file.write_text(f"{code}\n", encoding="utf-8")
        content = log.read_text(encoding="utf-8", errors="replace") if log.is_file() else ""
        result = evaluate(content, stamp, require_l130=(seed == 0))
        row = {"seed": seed, "process_exit": code, "source_stamp": stamp,
               "elapsed_s": round(time.monotonic() - started, 3), "resumed": resumed,
               "step3_passed": result["step3_passed"], **result,
               "log": str(log.relative_to(ROOT)), "layout": str(layout.relative_to(ROOT))}
        rows.append(row)
        (OUT / "runs-partial.json").write_text(json.dumps(rows, indent=2) + "\n")
        print("[L176SeedRun] " + json.dumps(row, sort_keys=True), flush=True)
    passed = sum(bool(row["step3_passed"]) for row in rows)
    summary = {"source_stamp": stamp, "seeds": list(range(args.seed_start, args.seed_stop)),
               "step3_passed": passed, "step3_failed": len(rows) - passed, "rows": rows}
    (OUT / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
    print(f"[L176SeedSummary] step3={passed}/{len(rows)}", flush=True)
    return 0 if passed == len(rows) else 1


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