wip(assembler): round L21 -- resolved strate layout snapshot + validator (no world change)

Core ResolvedLayoutSnapshotV1 (canonical JSON, stable identity): per strate stable ID, generator,
recipe, interior bounds in metres, ceiling/bedrock bands as Core really applies them, transition
mode/width, dispatch provenance. Validator: compat mode reports today's deviations; strict mode
enforces ceiling -> content -> bedrock -> next ceiling (owner rule), two separate bands, nominal
contiguity, interpenetration only within the declared EVoxelStrateTransition width; 7 negative
controls rejected. Findings: SW built at -120..-632 m although the params say -60 m; a 60 m unowned
gap filled by analytic FI fallback; no active bedrock bands; transitions ignored by Core.
All L19 hashes unchanged (28 + 14 crossing runs). The Unreal adapter exports resolved_layout
(uncompiled here).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CSFYgsdW8ejZSyiMf4qRSX

Standalone/Research/WorldAssemblerM3/Linux21/RunEvidenceL21.py

@@ -0,0 +1,165 @@
+#!/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)