#!/usr/bin/env python3 """Run L146's composed liminal audits and the worker/compiler identity matrix.""" from __future__ import annotations import argparse import json import os from pathlib import Path import re import subprocess import sys import time ROOT = Path(__file__).resolve().parents[4] OUT = ROOT / "Saved/Linux/l146" FIXTURE = OUT / "fixtures/core-liminal-field-3-strata.json" LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"] AUDIT_FLAGS = ("--audit-resolved-layout-passages", "--audit-liminal-road-walk", "--audit-passage-tile-seams") VARIANTS = ("flatField-monolith", "periodicHills", "road-over-cloud-floor", "pillar-grid") PASSAGE_RE = re.compile(r"^\[ResolvedPassageAudit\].*walkability=PASS$", re.M) SHELL_RE = re.compile(r"^\[PassageShellAudit\].*status=PASS$", re.M) sys.path.insert(0, str(ROOT / "Standalone/Tools/verify")) from runner import current_stamp, first_stamp def parse_seeds(value: str) -> list[int]: result: set[int] = set() for token in value.split(","): token = token.strip() if not token: continue if "-" in token: first, last = (int(part) for part in token.split("-", 1)) if first > last: raise argparse.ArgumentTypeError(f"reversed seed range: {token}") result.update(range(first, last + 1)) else: result.add(int(token)) if not result or min(result) < 0: raise argparse.ArgumentTypeError("seeds must be nonnegative and nonempty") return sorted(result) def validate_audit(output: str, seed: int, compiler: str, workers: int) -> None: checks = { "landing-pad negative control": "[LandingPadNegativeControl] uncertified_rejected=true status=PASS" in output, "passage walk audits": len(PASSAGE_RE.findall(output)) == 2, "passage interior capsule audit": len(re.findall( r"^\[ResolvedPassageAudit\].*passage_interior_capsule_samples=[1-9]\d* " r"passage_interior_solid_samples=0 interior_air=PASS .*walkability=PASS$", output, re.M)) == 2, "passage interior negative controls": len(re.findall( r"^\[ResolvedPassageInteriorNegativeControl\].*interior_solid_samples=[1-9]\d* " r"detected=true status=PASS$", output, re.M)) == 2, "passage interior audit summary": bool(re.search( r"^\[PassageInteriorCapsuleAuditSummary\].*status=PASS$", output, re.M)), "passage shell audits": len(SHELL_RE.findall(output)) == 2, "passage ripple audits": len(re.findall( r"^\[PassageRippleAudit\].*status=PASS$", output, re.M)) == 2, "masked ripple sampling controls": len(re.findall( r"^\[PassageRippleSamplingControl\].*status=PASS$", output, re.M)) == 2, "passage shell summary": bool(re.search( r"^\[PassageShellAuditSummary\].*status=PASS$", output, re.M)), "liminal walk audit": bool(re.search( r"^\[LiminalLayerWalkAudit\].*status=PASS$", output, re.M)), "liminal negative control": "[LiminalLayerNegativeControl] altered_departure_rejected=true status=PASS" in output, "passage summary": bool(re.search( r"^\[ResolvedPassageAuditSummary\].*status=PASS$", output, re.M)), "seam audit": bool(re.search( r"^\[PassageTileSeamAudit\].*status=PASS$", output, re.M)), } failed = [name for name, passed in checks.items() if not passed] if failed: raise RuntimeError( f"seed={seed} compiler={compiler} workers={workers} failed checks: " + ", ".join(failed)) def run_audit(seed: int, compiler: str, workers: int, *, audit: bool = True) -> dict: binary = OUT / "fast/bin" / f"VoxelForgeCoreHeadless-fast-{compiler}" if not binary.is_file(): raise RuntimeError(f"missing binary: {binary}") expected_stamp = current_stamp() probe = subprocess.run([str(binary), "--print-source-stamp"], cwd=ROOT, capture_output=True, text=True, timeout=15, check=False) if probe.returncode != 0 or first_stamp(probe.stdout) != expected_stamp: raise RuntimeError(f"stale executable: {binary}") tag = f"seed{seed:02d}-{compiler}-w{workers}" log = OUT / "logs" / f"audit-{tag}.log" manifest = OUT / "manifests" / f"audit-{tag}.json" layout = OUT / "layouts" / f"audit-{tag}.json" manifest.parent.mkdir(parents=True, exist_ok=True) layout.parent.mkdir(parents=True, exist_ok=True) command = [str(binary), str(FIXTURE), f"--workers={workers}", "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--force-generate", "--clip-radius=1", "--max-clip-level=0", "--max-level=0", "--render-distance-chunks=0", "--camera=0,0,-540,0,0", f"--seed={seed}", f"--world-assembler-manifest-output={manifest}", f"--resolved-layout-output={layout}"] if audit: command.extend(AUDIT_FLAGS) env = os.environ.copy() env["VOXELFORGE_VERIFY_OUT"] = str(OUT) env["VOXELFORGE_OWNER_DATA"] = str(ROOT / "Standalone/data/owner_world.json") env["LP_NUM_THREADS"] = "12" env["XDG_CACHE_HOME"] = str(OUT / "xdg-cache") started = time.monotonic() with log.open("w", encoding="utf-8") as stream: result = subprocess.run([*LIMIT, "timeout", "1740s", *command], cwd=ROOT, env=env, stdout=stream, stderr=subprocess.STDOUT, timeout=1745, check=False) output = log.read_text(encoding="utf-8", errors="replace") if result.returncode != 0: raise RuntimeError(f"{tag} exit={result.returncode}; see {log}") if first_stamp(output) != expected_stamp or current_stamp() != expected_stamp: raise RuntimeError(f"{tag} source stamp changed during run; see {log}") if audit: validate_audit(output, seed, compiler, workers) if not manifest.is_file() or not layout.is_file(): raise RuntimeError(f"{tag} omitted canonical manifest or layout") manifest_value = json.loads(manifest.read_text(encoding="utf-8")) layout_value = json.loads(layout.read_text(encoding="utf-8")) identity = { "seed": seed, "variant": VARIANTS[seed % len(VARIANTS)], "compiler": compiler, "workers": workers, "manifest_hash": manifest_value["content_hash"], "resolved_layout_identity": layout_value["stable_identity"], "log": str(log.relative_to(ROOT)), "elapsed_s": round(time.monotonic() - started, 3), "audited": audit, } print(f"[L146AuditRun] seed={seed} variant={identity['variant']} compiler={compiler} " f"workers={workers} manifest={identity['manifest_hash']} " f"layout={identity['resolved_layout_identity']} " f"elapsed_s={identity['elapsed_s']:.3f} status=PASS", flush=True) return identity def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--seeds", default="0-31", type=parse_seeds, help="audit seeds, e.g. 0-31 or 0,1,2,3") parser.add_argument("--determinism", action="store_true", help="also compare seed 0 on Clang/GCC at W1/W8/W12") parser.add_argument("--only-determinism", action="store_true", help="run only the Clang/GCC seed-0 W1/W8/W12 identity matrix") args = parser.parse_args() if not FIXTURE.is_file(): raise SystemExit(f"missing fixture: {FIXTURE}; build the L145 liminal fixture first") records = [] if not args.only_determinism: for seed in args.seeds: records.append(run_audit(seed, "clang", 1)) if args.determinism or args.only_determinism: matrix = [run_audit(0, compiler, workers) for compiler in ("clang", "gcc") for workers in (1, 8, 12)] identities = {(row["manifest_hash"], row["resolved_layout_identity"]) for row in matrix} if len(identities) != 1: raise RuntimeError(f"Clang/GCC W1/W8/W12 identity mismatch: {matrix}") records.extend(matrix) summary = { "audit_seeds": args.seeds if not args.only_determinism else [], "variant_counts": { name: sum(1 for seed in args.seeds if seed % len(VARIANTS) == index) for index, name in enumerate(VARIANTS) } if not args.only_determinism else {}, "records": records, "determinism_pass": bool(args.determinism or args.only_determinism), "status": "PASS", } summary_path = OUT / "l146-audit-matrix-summary.json" summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") print(f"[L146AuditMatrix] records={len(records)} summary={summary_path} status=PASS", flush=True) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as error: print(f"[L146AuditMatrix] status=FAIL error={error}", file=sys.stderr, flush=True) raise SystemExit(1)