#!/usr/bin/env python3 """Capture composed passage, route, shell, and seam attribution for seeds 0..15.""" from __future__ import annotations import argparse import json import re import subprocess import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] DEFAULT_BIN = ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-l174-step0-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) GATES = { "passages": r"^\[ResolvedPassageAuditSummary\].*status=PASS$", "seams": r"^\[PassageTileSeamAudit\].*status=PASS$", "shell": r"^\[PassageShellAuditSummary\].*status=PASS$", "naturalness": r"^\[PassageNaturalnessSummary\].*status=PASS$", "mouth_profile": r"^\[PassageMouthProfileSummary\].*status=PASS$", "ripple": r"^\[PassageRippleAuditSummary\].*status=PASS$", "passage_interior": r"^\[PassageInteriorCapsuleAuditSummary\].*status=PASS$", "self_overlap": r"^\[PassageTubeSelfOverlapAuditSummary\].*status=PASS$", } def evaluate(content: str, expected_stamp: str, layout_path: Path, require_end_runs: bool = False) -> tuple[dict, dict, str | None]: stamp = STAMP_RE.search(content) checks = {name: re.search(pattern, content, re.M) is not None for name, pattern in GATES.items()} if require_end_runs: checks["end_runs"] = re.search( r"^\[PassageEndRunAuditSummary\].*status=PASS$", content, re.M) is not None checks["source_stamp"] = stamp is not None and stamp.group(1) == expected_stamp for kind, strate in (("tunnel_network", 2), ("vertical_shafts", 3)): route = re.compile( rf"^\[ResolvedStrateRouteAudit\] strate={strate} .*unsupported=0 " rf"grade_failures=0 .*status=PASS$", re.M) checks[f"{kind}_route"] = route.search(content) is not None checks[f"{kind}_blocked_route_control"] = re.search( rf"^\[{kind.title().replace('_', '')}BlockedRouteControl\].*" r"expected=FAIL status=PASS$", content, re.M) is not None checks["passage_count"] = len(re.findall( r"^\[ResolvedPassageAudit\] .*walkability=PASS$", content, re.M)) == 3 checks["negative_controls"] = ( len(re.findall(r"^\[ResolvedPassageInteriorNegativeControl\].*" r"detected=true status=PASS$", content, re.M)) == 3 and len(re.findall(r"^\[ResolvedPassageBlockedControl\].*detected=true$", content, re.M)) == 3 ) attribution = { "seam_samples_by_tag": { tag: len(re.findall(rf"^\[PassageTileSeamSample\] tag={re.escape(tag)} ", content, re.M)) for tag in ("disc", "rim", "end run", "body", "route") }, "landing_support_failures_upper": len(re.findall( r"^\[PassageLandingSupportFailure\].*side=upper", content, re.M)), "landing_support_failures_lower": len(re.findall( r"^\[PassageLandingSupportFailure\].*side=lower", content, re.M)), "unsupported_walk_samples": len(re.findall( r"^\[ResolvedWalkDispatch\] kind=unsupported", content, re.M)), "shell_floor_samples": len(re.findall( r"^\[PassageShellAuditSample\]", content, re.M)), "shell_wall_samples": len(re.findall( r"^\[PassageShellAuditSample\].*kind=wall_missing", content, re.M)), "tube_self_overlap_rows": len(re.findall( r"^\[PassageTubeSelfOverlapAudit\]", content, re.M)), "tube_self_overlap_pairs": sum(int(value) for value in re.findall( r"^\[PassageTubeSelfOverlapAudit\].*overlap_pairs=(\d+)", content, re.M)), "tube_end_turn_violations": sum(int(value) for value in re.findall( r"^\[PassageTubeSelfOverlapAudit\].*end_turn_violations=(\d+)", content, re.M)), "tube_landing_column_violations": sum(int(value) for value in re.findall( r"^\[PassageTubeSelfOverlapAudit\].*landing_column_violations=(\d+)", content, re.M)), "end_run_failures": len(re.findall( r"^\[PassageEndRunAudit\].*status=FAIL$", content, re.M)), "end_run_measured": len(re.findall( r"^\[PassageEndRunAudit\].*measured_start_m=(?!0\.000)" r".*measured_end_m=(?!0\.000)", content, re.M)), } seam_summary = re.search(r"^\[PassageTileSeamAudit\].*$", content, re.M) if seam_summary: for key in ("candidates", "examined", "pairs", "lod0_pairs", "lod0_samples", "lod1_pairs", "lod1_samples", "landing_discs", "end_runs", "lod0_disc_samples", "lod1_disc_samples", "lod0_rim_samples", "lod1_rim_samples", "lod0_end_run_samples", "lod1_end_run_samples", "exempt_pairs", "mismatches"): match = re.search(rf"\b{key}=(\d+)", seam_summary.group(0)) attribution[f"seam_{key}"] = int(match.group(1)) if match else None for key in ("landing_discs", "end_runs"): match = re.search(rf"\b{key}=(\d+)/(\d+)", seam_summary.group(0)) if match: attribution[f"seam_{key}_covered"] = int(match.group(1)) attribution[f"seam_{key}_required"] = int(match.group(2)) attribution["seam_uncovered_disc_regions"] = len(re.findall( r"^\[PassageTileSeamCoverage\].*\bdisc_status=FAIL\b", content, re.M)) attribution["seam_uncovered_end_run_regions"] = len(re.findall( r"^\[PassageTileSeamCoverage\].*\bend_run_status=FAIL\b", content, re.M)) attribution["seam_uncovered_regions"] = ( attribution["seam_uncovered_disc_regions"] +attribution["seam_uncovered_end_run_regions"]) passage_summary = re.search(r"^\[ResolvedPassageAuditSummary\].*$", content, re.M) if passage_summary: for key in ("support_probes", "exempt_support_probes", "unsupported"): match = re.search(rf"\b{key}=(\d+)", passage_summary.group(0)) if match: attribution[key] = int(match.group(1)) identity = None try: if layout_path.is_file(): identity = json.loads(layout_path.read_text(encoding="utf-8")).get( "stable_identity") except (OSError, json.JSONDecodeError): pass checks["layout_output"] = identity is not None return checks, attribution, identity def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--binary", type=Path, default=DEFAULT_BIN) parser.add_argument("--out", type=Path, default=ROOT / "Saved/Linux/l174/rebaseline-l173/seeds") parser.add_argument("--seed-start", type=int, default=0) parser.add_argument("--seed-stop", type=int, default=16, help="exclusive seed bound") parser.add_argument("--expected-stamp", type=str, help="source stamp embedded in an earlier binary") parser.add_argument("--require-end-runs", action="store_true", help="require the L174 end-run audit (use for post-Step-1 sweeps)") parser.add_argument("--no-resume", action="store_true", help="rerun even when both the log and layout already exist") args = parser.parse_args() if not args.binary.is_file() or not DATA.is_file(): raise SystemExit(f"missing binary or owner fixture: {args.binary}") args.binary = args.binary.resolve() args.out = args.out.resolve() args.out.mkdir(parents=True, exist_ok=True) stamp = args.expected_stamp or subprocess.check_output( [sys.executable, str(STAMP_TOOL), "stamp", "--root", str(ROOT)], text=True).strip() results = [] partial = args.out / "runs-partial.json" for seed in range(args.seed_start, args.seed_stop): log = args.out / f"seed{seed}.log" layout = args.out / f"seed{seed}.layout.json" exit_path = args.out / f"seed{seed}.exit" command = [ "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", "timeout", "1740s", str(args.binary), 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}", ] started = time.monotonic() resumed = False if not args.no_resume and log.is_file() and layout.is_file(): content = log.read_text(encoding="utf-8", errors="replace") if exit_path.is_file(): code: int | str = int(exit_path.read_text().strip()) elif ("[PipelineExactQueryHash]" in content and "[ResolvedPassageAuditSummary]" in content): code = 10 if re.search(r"^\[[^]]*(?:Summary|Audit)\].*status=FAIL$", content, re.M) else 0 else: code = "NO_EXIT_RECORD" resumed = code != "NO_EXIT_RECORD" if not resumed: try: with log.open("w", encoding="utf-8") as output: proc = subprocess.run(command, cwd=ROOT, stdout=output, stderr=subprocess.STDOUT, check=False, timeout=1740) code = proc.returncode except subprocess.TimeoutExpired: code = "OUTER_TIMEOUT" if isinstance(code, int): exit_path.write_text(str(code) + "\n", encoding="utf-8") content = log.read_text(encoding="utf-8", errors="replace") if log.exists() else "" checks, attribution, identity = evaluate( content, stamp, layout, args.require_end_runs) passed = code == 0 and all(checks.values()) failed = [name for name, ok in checks.items() if not ok] row = { "seed": seed, "verdict": "PASS" if passed else "FAIL", "exit": code, "failed_gates": failed, "checks": checks, "attribution": attribution, "source_stamp": stamp, "layout_identity": identity, "elapsed_s": round(time.monotonic() - started, 3), "log": str(log.relative_to(ROOT)), "layout": str(layout.relative_to(ROOT)), "resumed": resumed, } results.append(row) partial.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") print("[L174CaveSeedSweep] " + json.dumps(row, sort_keys=True), flush=True) summary = { "source_stamp": stamp, "seeds": list(range(args.seed_start, args.seed_stop)), "passed": sum(row["verdict"] == "PASS" for row in results), "failed": sum(row["verdict"] != "PASS" for row in results), "results": results, } (args.out / "summary.json").write_text( json.dumps(summary, indent=2) + "\n", encoding="utf-8") print(f"[L174CaveSeedSweepSummary] passed={summary['passed']}/{len(results)} " f"failed={summary['failed']}", flush=True) return 0 if summary["failed"] == 0 else 1 if __name__ == "__main__": raise SystemExit(main())