VoxelForge/Standalone/Research/WorldAssemblerM3/Linux21/RunEvidenceL21.py

#!/usr/bin/env python3
"""Constrained L21 snapshot, output-parity, crossing, and off-cost evidence."""
import os
import re
import signal
import statistics
import subprocess
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/linux21"
OUT.mkdir(parents=True, exist_ok=True)
if not os.sched_getaffinity(0).issubset(set(range(12))):
    raise SystemExit("run under taskset -c 0-11")

BIN = {
    "clang-off": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux21-clang",
    "gcc-default": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux21-gcc",
}
OLD_CLANG = ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux19-clang"
OLD_GCC = ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux19-gcc"
DATA = ROOT / "Standalone/data/owner_world.json"
HASH_ROW = re.compile(r"^\[Pipeline(?:FullDetailHash|WholeViewHash|LODHash|ExactQueryHash)\].*$", re.M)
LODS = [(1, 1), (8, 1), (12, 1), (12, 2), (12, 3), (12, 4), (12, 5)]
RECIPES = {
    "surface-world": "0,-35,-487.94,0,-22",
    "floating-islands": "-22,-19,-22.8,0,-4",
}
results = []


def launch(name, argv, timeout=900):
    logfile = OUT / f"{name}.log"
    command = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", *map(str, argv)]
    started = time.monotonic()
    peak_tree = 0
    peak_process = 0
    timed_out = False
    with logfile.open("w", encoding="utf-8") as stream:
        process = subprocess.Popen(command, stdout=stream, stderr=subprocess.STDOUT,
                                   start_new_session=True)
        while process.poll() is None:
            pending, seen, total = [process.pid], set(), 0
            while pending:
                pid = pending.pop()
                if pid in seen:
                    continue
                seen.add(pid)
                try:
                    pending.extend(int(x) for x in Path(f"/proc/{pid}/task/{pid}/children").read_text().split())
                    values = {}
                    for line in Path(f"/proc/{pid}/status").read_text().splitlines():
                        if line.startswith(("VmRSS:", "VmHWM:")):
                            key, amount, *_ = line.split()
                            values[key] = int(amount)
                    total += values.get("VmRSS:", 0)
                    peak_process = max(peak_process, values.get("VmRSS:", 0), values.get("VmHWM:", 0))
                except (FileNotFoundError, PermissionError, ProcessLookupError, ValueError):
                    pass
            peak_tree = max(peak_tree, total)
            if time.monotonic() - started > timeout:
                os.killpg(process.pid, signal.SIGTERM)
                try:
                    process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    os.killpg(process.pid, signal.SIGKILL)
                    process.wait()
                timed_out = True
                break
            time.sleep(0.1)
        code = process.wait()
    elapsed = time.monotonic() - started
    rows = HASH_ROW.findall(logfile.read_text(encoding="utf-8", errors="replace"))
    results.append((name, code, elapsed, peak_process, peak_tree, timed_out, len(rows)))
    print(f"[L21Run] {name} exit={code} seconds={elapsed:.3f} peak_process_rss_kib={peak_process} "
          f"peak_tree_rss_kib={peak_tree} hash_rows={len(rows)} timed_out={str(timed_out).lower()}", flush=True)
    return logfile


def base_args(binary, recipe, workers, hash_output=False, crossing=False):
    args = [binary, DATA, "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
            "--force-generate", "--clip-radius=3", "--max-clip-level=8", f"--workers={workers}",
            f"--camera={RECIPES[recipe]}", f"--recipe={recipe}"]
    if hash_output:
        args.append("--stream-hash")
    if crossing:
        args += ["--camera-step=24,0,0", "--region-crossing"]
    return args


# Plan-time identities and canonical serialization at W1/W8/W12, both field recipes.
snapshots = {}
for recipe in RECIPES:
    for workers, repeat in [(1, 1), (8, 1), (12, 1)]:
        path = OUT / f"snapshot-{recipe}-w{workers}-r{repeat}.json"
        args = [BIN["clang-off"], DATA, "--world-assembler-prototype", "--stream-benchmark",
                "--stream-pipeline", f"--recipe={recipe}", f"--workers={workers}",
                f"--resolved-layout-output={path}", "--resolved-layout-only"]
        log = launch(f"snapshot-{recipe}-w{workers}-r{repeat}", args)
        if results[-1][1] != 0:
            raise SystemExit(f"snapshot plan failed: {log}")
        snapshots[(recipe, workers)] = path.read_bytes()
    if len({snapshots[(recipe, workers)] for workers in (1, 8, 12)}) != 1:
        raise SystemExit(f"snapshot serialization changed with worker count: {recipe}")
    print(f"[L21SnapshotIdentity] recipe={recipe} worker_counts=1,8,12 canonical_bytes=identical status=PASS", flush=True)

# Full matrix. Every row is compared against the L19 W12x5 reference.
for recipe in RECIPES:
    for compiler, binary in BIN.items():
        for workers, repeat in LODS:
            name = f"matrix-{recipe}-{compiler}-w{workers}-r{repeat}"
            log = launch(name, base_args(binary, recipe, workers, hash_output=True))
            rows = HASH_ROW.findall(log.read_text(encoding="utf-8", errors="replace"))
            reference_name = f"{recipe}-{compiler}-w12-r5.log"
            reference = ROOT / "Saved/Linux/l19-matrix" / reference_name
            expected = HASH_ROW.findall(reference.read_text(encoding="utf-8", errors="replace"))
            if results[-1][1] != 0 or len(rows) != 12 or rows != expected:
                print(f"[L21HashMismatch] {name} expected_rows={len(expected)} actual_rows={len(rows)}", flush=True)
                raise SystemExit(1)

# The crossing matrix retains the 24 m two-view query proof on both compilers.
for compiler, binary in BIN.items():
    for workers, repeat in LODS:
        name = f"crossing-{compiler}-w{workers}-r{repeat}"
        log = launch(name, base_args(binary, "surface-world", workers, hash_output=True, crossing=True))
        rows = HASH_ROW.findall(log.read_text(encoding="utf-8", errors="replace"))
        reference = ROOT / "Saved/Linux/l19-crossing" / f"{compiler}-w12-r5.log"
        expected = HASH_ROW.findall(reference.read_text(encoding="utf-8", errors="replace"))
        if results[-1][1] != 0 or len(rows) != 12 or rows != expected:
            print(f"[L21CrossingHashMismatch] {name} expected_rows={len(expected)} actual_rows={len(rows)}", flush=True)
            raise SystemExit(1)

# Five interleaved W12 pairs per recipe, hashing disabled. Alternate order by pair to reduce drift.
off_pairs = {recipe: {"old": [], "new": []} for recipe in RECIPES}
for recipe in RECIPES:
    for pair in range(1, 6):
        order = [("old", OLD_CLANG), ("new", BIN["clang-off"])] if pair % 2 else [("new", BIN["clang-off"]), ("old", OLD_CLANG)]
        for label, binary in order:
            name = f"offcost-{recipe}-pair{pair}-{label}"
            log = launch(name, base_args(binary, recipe, 12, hash_output=False), timeout=900)
            if results[-1][1] != 0:
                raise SystemExit(f"hashing-off cost run failed: {log}")
            off_pairs[recipe][label].append(results[-1][2])
        print(f"[L21OffCostPair] recipe={recipe} pair={pair} old_seconds={off_pairs[recipe]['old'][-1]:.3f} "
              f"new_seconds={off_pairs[recipe]['new'][-1]:.3f}", flush=True)
for recipe, pair in off_pairs.items():
    old = statistics.median(pair["old"])
    new = statistics.median(pair["new"])
    print(f"[L21OffCostSummary] recipe={recipe} pairs=5 median_old_seconds={old:.3f} "
          f"median_new_seconds={new:.3f} delta_percent={(new-old)/old*100.0:.2f}", flush=True)

# Query the real compiled Core field at every distinct recorded stratum/band endpoint.
for recipe in RECIPES:
    name = f"boundary-probes-{recipe}"
    args = base_args(BIN["clang-off"], recipe, 12, hash_output=False)
    args += ["--audit-resolved-layout-boundaries"]
    log = launch(name, args)
    if results[-1][1] != 0:
        raise SystemExit(f"Core boundary probes failed: {log}")

print(f"[L21EvidenceSummary] snapshots=6 matrix=28 crossing=14 off_cost_pairs=10 boundary_audits=2 status=PASS", flush=True)
for name, code, elapsed, rss, tree, timed_out, rows in results:
    print(f"[L21RunDetail] {name} exit={code} seconds={elapsed:.3f} peak_process_rss_kib={rss} "
          f"peak_tree_rss_kib={tree} hash_rows={rows} timed_out={str(timed_out).lower()}", flush=True)