VoxelForge/Standalone/Research/WorldAssemblerM3/Linux139/RunL139SeedSweep.py

#!/usr/bin/env python3
"""Audit the composed owner world, including the basal Vertical Shafts route."""
from __future__ import annotations

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


ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l139/seeds"
BIN = ROOT / "Saved/Linux/l139/bin/VoxelForgeCoreHeadless-l139-clang"
DATA = ROOT / "Standalone/data/owner_world.json"
STAMP = "b8603b6cb3975ca0f9669759e4e2464561c557fdb7aa6836acef123ecf0b99d1"
STAMP_RE = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]+)$", re.M)
REQUIRED = {
    "passages": re.compile(r"^\[ResolvedPassageAuditSummary\].*status=PASS$", re.M),
    "route": re.compile(
        r"^\[ResolvedStrateRouteAudit\] strate=3 .*unsupported=0 grade_failures=0 .*status=PASS$",
        re.M),
    "seams": re.compile(r"^\[PassageTileSeamAudit\].*status=PASS$", re.M),
    "shell": re.compile(r"^\[PassageShellAuditSummary\].*status=PASS$", re.M),
    "naturalness": re.compile(r"^\[PassageNaturalnessSummary\].*status=PASS$", re.M),
    "mouth_profile": re.compile(r"^\[PassageMouthProfileSummary\].*status=PASS$", re.M),
    "ripple": re.compile(r"^\[PassageRippleAuditSummary\].*status=PASS$", re.M),
    "shaft_dispatch": re.compile(
        r"^\[DensityDispatch\] strate=3 generator=VerticalShafts "
        r"recipe=Core\.Caves\.PrimitiveField version=1 "
        r"bound_path=Caves-native adapter=FCavesFieldAdapter$", re.M),
    "shaft_blocked_route": re.compile(
        r"^\[VerticalShaftsBlockedRouteControl\].*expected=FAIL status=PASS$", re.M),
}


def main() -> int:
    if not BIN.is_file() or not DATA.is_file():
        raise SystemExit("missing L139 Clang binary or composed owner fixture")
    OUT.mkdir(parents=True, exist_ok=True)
    results = []
    partial = OUT / "runs-partial.json"
    if partial.is_file():
        for prior in json.loads(partial.read_text(encoding="utf-8")):
            seed = len(results)
            if prior.get("seed") != seed:
                break
            log = ROOT / prior.get("log", "")
            layout = ROOT / prior.get("layout", "")
            if not log.is_file() or not layout.is_file():
                break
            content = log.read_text(encoding="utf-8", errors="replace")
            stamp = STAMP_RE.search(content)
            checks = {name: bool(pattern.search(content))
                      for name, pattern in REQUIRED.items()}
            checks["source_stamp"] = stamp is not None and stamp.group(1) == STAMP
            checks["blocked_passage_controls"] = len(re.findall(
                r"^\[ResolvedPassageBlockedControl\].*detected=true$", content, re.M)) == 3
            checks["passage_count"] = len(re.findall(
                r"^\[ResolvedPassageAudit\].*walkability=PASS$", content, re.M)) == 3
            passed = prior.get("exit") == 0 and all(checks.values())
            if checks != prior.get("checks") or passed != prior.get("passed"):
                break
            results.append(prior)
        print(f"[L139SeedSweepResume] reused={len(results)} source_stamp={STAMP}",
              flush=True)
    for seed in range(len(results), 16):
        log = OUT / f"seed{seed}.log"
        layout = OUT / f"seed{seed}.json"
        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}"]
        started = time.monotonic()
        with log.open("w", encoding="utf-8") as output:
            proc = subprocess.run(command, cwd=ROOT, stdout=output,
                                  stderr=subprocess.STDOUT, check=False, timeout=1740)
        content = log.read_text(encoding="utf-8", errors="replace")
        stamp = STAMP_RE.search(content)
        checks = {name: bool(pattern.search(content)) for name, pattern in REQUIRED.items()}
        checks["source_stamp"] = stamp is not None and stamp.group(1) == STAMP
        checks["blocked_passage_controls"] = len(re.findall(
            r"^\[ResolvedPassageBlockedControl\].*detected=true$", content, re.M)) == 3
        checks["passage_count"] = len(re.findall(
            r"^\[ResolvedPassageAudit\].*walkability=PASS$", content, re.M)) == 3
        passed = proc.returncode == 0 and all(checks.values())
        row = {"seed": seed, "exit": proc.returncode, "passed": passed,
               "checks": checks, "elapsed_s": round(time.monotonic() - started, 3),
               "log": str(log.relative_to(ROOT)), "layout": str(layout.relative_to(ROOT))}
        results.append(row)
        print("[L139SeedSweep] " + json.dumps(row, sort_keys=True), flush=True)
        partial.write_text(json.dumps(results, indent=2) + "\n")
    passed_count = sum(row["passed"] for row in results)
    failed_count = len(results) - passed_count
    summary = {"source_stamp": STAMP, "seeds": list(range(len(results))),
               "passed": passed_count, "failed": failed_count, "results": results}
    (OUT / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
    print(f"[L139SeedSweepSummary] passed={passed_count}/16 failed={failed_count} "
          f"status={'PASS' if len(results)==16 and failed_count==0 else 'FAIL'}")
    return 0 if len(results) == 16 and failed_count == 0 else 1


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