Files
VoxelForge/Standalone/Research/WorldAssemblerM3/Linux37/RunRingAuditsL37.py
T
Fr0zkaandClaude Opus 5.5 d970d75be8 streaming: deeper ready queue + startup apply cap; tile-local mouth-support culling (L72)
- Unreal: ready queue depth 8 -> 16 (cap 4 worker-sets / 64), startup mesh-apply cap
  12 -> 24 within the unchanged 14 ms budget. Owner SW arrival, same 2,850 keys:
  initial set ready 4.78 -> 3.74 s, queue p95 123 -> 35 ms, request-to-apply p95
  139 -> 45 ms; walkable gate 1.07 -> 1.11 s; per-frame apply p95 0.73 ms.
- Core composer: mouth-support segments filtered by tile bounds, mouth relief result reused
  (route checks -80/-87% in passage tiles). Hashes identical W1/W8/W12, Clang and GCC.
- Opt-in per-tile composer counters. Verify control meshes stored as .xz (78.5 -> 18.4 MB).
- Fast tier 269 PASS.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
2026-10-27 08:25:14 +01:00

210 lines
10 KiB
Python

#!/usr/bin/env python3
"""Export v24 ring meshes at matching LOD windows and retain old controls."""
import csv
import argparse
import re
import subprocess
import sys
import time
import os
from pathlib import Path
ROOT = Path(__file__).resolve().parents[4]
OUT = Path(os.environ.get("VOXELFORGE_VERIFY_OUT", ROOT / "Saved/Linux/l37"))
BASELINES = OUT / "baselines"
BIN = Path(os.environ.get("VOXELFORGE_VERIFY_HEADLESS",
OUT / "bin/VoxelForgeCoreHeadless-l37-final-clang"))
OWNER = Path(os.environ.get("VOXELFORGE_OWNER_DATA",
ROOT / "Standalone/data/owner_world.json"))
TERRACE = ROOT / "Standalone/Research/WorldAssemblerM3/Linux32/AuditRingTerraceL32.py"
RING_SET = ROOT / "Standalone/Research/WorldAssemblerM3/Linux32/AuditRingSetL32.py"
STAMP = re.compile(
r"^\[WorldAssemblerRecipeAudit\] requested=([^ ]+) actual=([^ ]+).*status=PASS$", re.M)
EXPORT = re.compile(r"^\[MeshSceneExport\].*status=PASS", re.M)
TERRACE_RESULT = re.compile(r"^\[RingTerraceAudit\].*status=(PASS|FAIL)$", re.M)
SET_RESULT = re.compile(r"^\[RingSetAudit\].*status=(PASS|FAIL)$", re.M)
rows = []
failed = False
PARSER = argparse.ArgumentParser(description=__doc__)
PARSER.add_argument("--lod0-only", action="store_true")
PARSER.add_argument("--scene", choices=("fi-floor", "sw-ceiling", "all"), default="all")
PARSER.add_argument("--skip-controls", action="store_true")
OPTIONS = PARSER.parse_args()
sys.path.insert(0, str(ROOT / "Standalone/Tools/verify"))
from runner import current_stamp, first_stamp, bounded_run
FI_CENTERS = ((56, -72), (72, -72), (56, -88), (40, -72), (24, 24),
(104, 24), (72, -88), (120, 24), (104, 8), (24, -24))
SW_CENTERS = ((-136, 136), (-56, 152), (-152, 120), (-136, 120),
(-152, 104), (-136, 104), (-40, 152))
LOD_CENTERS = {
"fi-floor": {
2: (16, -144), 3: (224, -160), 4: (-448, 448),
5: (-128, -640), 6: (-1792, 1280), 7: (3584, 1536),
8: (3072, 7168),
},
"sw-ceiling": {
2: (16, -144), 3: (224, -160), 4: (-448, 448),
5: (-128, -640), 6: (-1792, 1280), 7: (2560, -512),
8: (3072, 7168),
},
}
def run(name, command):
log = OUT / f"ring-{name}.log"
wrapped = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
"timeout", "1800s", *map(str, command)]
stamped = bool(command and Path(command[0]) == BIN)
expected_stamp = current_stamp() if stamped else None
if stamped:
probe = bounded_run([str(BIN), "--print-source-stamp"], timeout=15)
if (probe is None or probe.returncode != 0
or first_stamp(probe.stdout or "") != expected_stamp):
raise SystemExit(f"{name} refused an unstamped or stale executable: {BIN}")
started = time.monotonic()
try:
with log.open("w", encoding="utf-8") as stream:
result = subprocess.run(wrapped, cwd=ROOT, stdout=stream,
stderr=subprocess.STDOUT, timeout=1800, check=False)
except subprocess.TimeoutExpired:
result = None
elapsed = time.monotonic() - started
output = log.read_text(encoding="utf-8", errors="replace")
if stamped and (first_stamp(output) != expected_stamp
or current_stamp() != expected_stamp):
raise SystemExit(f"{name} source stamp mismatch; see {log}")
code = result.returncode if result is not None else "TIMEOUT"
print(f"[L37RingStep] name={name} elapsed={elapsed:.1f}s exit={code} log={log}",
flush=True)
return result, output, elapsed
def export(label, recipe, max_level, camera):
mesh = OUT / f"ring-owner-{label}-v24.vfmesh"
layout = OUT / "layouts" / f"ring-layout-{label}.json"
layout.parent.mkdir(parents=True, exist_ok=True)
command = [BIN, OWNER, "--workers=12",
"--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
"--force-generate", "--clip-radius=3", "--max-clip-level=8",
f"--max-level={max_level}", f"--recipe={recipe}"]
if camera:
command.append(f"--camera={camera}")
command.extend((f"--mesh-export={mesh}", f"--resolved-layout-output={layout}"))
result, output, elapsed = run(f"export-{label}", command)
stamps = STAMP.findall(output)
ok = (result is not None and result.returncode == 0
and stamps == [(recipe, recipe)] and EXPORT.search(output) and mesh.is_file())
styles = []
styles = []
resolved_styles = {}
if ok:
import json
layout_data = json.loads(layout.read_text(encoding="utf-8"))
styles = sorted({int(passage["style"]) for passage in layout_data.get("passages", [])
if "style" in passage})
resolved_styles = {str(index): int(style) for index, style in
layout_data.get("resolved_strate_passage_styles", {}).items()}
ok = styles == [2] and resolved_styles == {"0": 2, "1": 1}
print(f"[L37RingExport] scene={label} recipe={recipe} stamp={stamps} "
f"passage_style={styles if mesh.exists() else []} resolved_strate_styles={resolved_styles} "
f"bytes={mesh.stat().st_size if mesh.exists() else 0} "
f"status={'PASS' if ok else 'FAIL'}", flush=True)
return mesh, elapsed, ok
def terrace(label, recipe, surface, mesh, lod, height_min, height_max,
center=(64, -32), radius=32):
result, output, elapsed = run(f"{label}-lod{lod}", ["python3", TERRACE, mesh,
f"--surface={surface}", "--center", str(center[0]), str(center[1]),
f"--radius={radius}",
f"--lod={lod}", f"--height-min-m={height_min}",
f"--height-max-m={height_max}"])
statuses = TERRACE_RESULT.findall(output)
ok = (result is not None and result.returncode == 0 and statuses == ["PASS"])
rows.append((label, recipe, surface, lod, elapsed, ok))
print(f"[L37RingAuditRun] scene={label} recipe={recipe} surface={surface} "
f"lod={lod} center={center} radius={radius} "
f"status={'PASS' if ok else 'FAIL'}", flush=True)
return ok
def lod1_set(label, recipe, surface, mesh, centers, height_min, height_max):
command = ["python3", RING_SET, mesh, f"--surface={surface}"]
for x, y in centers:
command.extend(("--center", str(x), str(y)))
command.extend(("--radius=8", "--lod=1", f"--height-min-m={height_min}",
f"--height-max-m={height_max}"))
result, output, elapsed = run(f"{label}-lod1-set", command)
statuses = SET_RESULT.findall(output)
ok = (result is not None and result.returncode == 0 and statuses == ["PASS"])
rows.append((label, recipe, surface, 1, elapsed, ok))
print(f"[L37RingAuditRun] scene={label} recipe={recipe} surface={surface} "
f"lod=1 windows={len(centers)} status={'PASS' if ok else 'FAIL'}",
flush=True)
return ok
SCENES = (
("fi-floor", "floating-islands", "floor", -122.0, -117.0,
"64,-32,-115,0,-45", "64,-32,-115,0,-45", FI_CENTERS),
("sw-ceiling", "surface-world", "ceiling", -132.0, -124.0,
"64,-32,-136,0,45", "-96,128,-136,0,45", SW_CENTERS))
selected_scenes = (SCENES if OPTIONS.scene == "all" else
tuple(scene for scene in SCENES if scene[0] == OPTIONS.scene))
for label, recipe, surface, window_min, window_max, camera0, camera1, centers in selected_scenes:
low_mesh, _, low_ok = export(f"{label}-lod0", recipe, 0, camera0)
failed |= not low_ok
if low_ok:
failed |= not terrace(label, recipe, surface, low_mesh, 0, window_min, window_max)
if not OPTIONS.lod0_only:
detail_mesh, _, detail_ok = export(f"{label}-lod1", recipe, 1, camera1)
failed |= not detail_ok
if detail_ok:
set_min, set_max = ((-121.625, -117.375) if label == "fi-floor"
else (-128.375, -124.125))
failed |= not lod1_set(label, recipe, surface, detail_mesh, centers,
set_min, set_max)
full_mesh, _, full_ok = export(f"{label}-lod2-8", recipe, 8, None)
failed |= not full_ok
if full_ok:
for lod in range(2, 9):
center = LOD_CENTERS[label][lod]
radius = 4 * (1 << lod)
failed |= not terrace(label, recipe, surface, full_mesh, lod,
window_min, window_max, center, radius)
# Keep the established L30 terrace and L31 flat-surface negative controls live.
controls = (
("l30-terrace", BASELINES / "ring/l32-fi-floor-l30-camera-negative.vfmesh",
-130.0, -110.0),
("l31-flat", BASELINES / "ring/l31-fi-floor-v19.vfmesh", -122.0, -117.0),
)
for name, mesh, height_min, height_max in (() if OPTIONS.skip_controls else controls):
result, output, _ = run(f"control-{name}", ["python3", TERRACE, mesh,
"--surface=floor", "--center", "64", "-32", "--radius=32", "--lod=0",
f"--height-min-m={height_min}", f"--height-max-m={height_max}"])
statuses = TERRACE_RESULT.findall(output)
detected = (result is not None and result.returncode == 1 and statuses == ["FAIL"])
failed |= not detected
print(f"[L37RingControl] name={name} detected={detected} "
f"status={'PASS' if detected else 'FAIL'}", flush=True)
scene_tag = OPTIONS.scene
lod_tag = "lod0" if OPTIONS.lod0_only else "lod0-8"
summary = OUT / f"ring-audits-{scene_tag}-{lod_tag}.csv"
with summary.open("w", newline="", encoding="utf-8") as stream:
writer = csv.writer(stream)
writer.writerow(("scene", "recipe", "surface", "lod", "elapsed_seconds", "status"))
writer.writerows((label, recipe, surface, lod, f"{elapsed:.3f}",
"PASS" if ok else "FAIL")
for label, recipe, surface, lod, elapsed, ok in rows)
expected_runs = (1 if OPTIONS.lod0_only else 9) * len(selected_scenes)
expected_controls = 0 if OPTIONS.skip_controls else len(controls)
print(f"[L37RingSummary] runs={len(rows)} pass={sum(row[-1] for row in rows)} "
f"fail={sum(not row[-1] for row in rows)} controls={expected_controls} "
f"status={'FAIL' if failed else 'PASS'} summary={summary}", flush=True)
raise SystemExit(1 if failed or len(rows) != expected_runs else 0)