Files
VoxelForge/Standalone/Research/WorldAssemblerM3/Linux143/CaptureL143.py
T
Fr0zkaandClaude Opus 5.5 a4e214561f proto: continuous swept road deck; coarse-LOD proxies keep thin pillars intact (L143)
Road: one swept-deck stamp on a shared cubic-Hermite centerline (stamp vocabulary V5), 376.8 m, max 8.5 deg, walk
audit clean, short-road control fails as expected. Pillars: regular fillPillar stamps get deterministic 24-sided
capped-cylinder proxies once the LOD step exceeds their radius or half-height (applies to all worlds). Owner hashes
unchanged (v227); five-fixture Clang/GCC determinism; fast tier green.

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

464 lines
23 KiB
Python

#!/usr/bin/env python3
"""Run L143 liminal determinism checks, walk audits and multi-LOD captures."""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import time
ROOT = Path(__file__).resolve().parents[4]
OUT_DEFAULT = Path(os.environ.get("VOXELFORGE_VERIFY_OUT", ROOT / "Saved/Linux/l143"))
CAPTURE_DEFAULT = Path(__file__).resolve().parent / "captures"
LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"]
Q_PER_METRE = 1024.0
FOG_COLOR = "0.66,0.73,0.77"
sys.path.insert(0, str(ROOT / "Standalone/Tools/verify"))
from runner import bounded_run, current_stamp, first_stamp
def fmt(values) -> str:
return ",".join(f"{value:.6f}" for value in values)
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def run_step(name: str, command: list[object], log: Path, *, env: dict[str, str],
stamped: bool = False, timeout_seconds: int = 1740,
expected_exit: int = 0) -> str:
expected_stamp = current_stamp() if stamped else None
if stamped:
binary = Path(str(command[0]))
probe = bounded_run([str(binary), "--print-source-stamp"], timeout=15)
if probe is None or probe.returncode != 0 or first_stamp(probe.stdout or "") != expected_stamp:
raise RuntimeError(f"{name} refused a stale executable: {binary}")
started = time.monotonic()
log.parent.mkdir(parents=True, exist_ok=True)
with log.open("w", encoding="utf-8") as stream:
result = subprocess.run([*LIMIT, f"timeout", f"{timeout_seconds}s",
*map(str, command)], cwd=ROOT, env=env,
stdout=stream, stderr=subprocess.STDOUT,
timeout=timeout_seconds + 5, check=False)
elapsed = time.monotonic() - started
output = log.read_text(encoding="utf-8", errors="replace")
print(f"[L143Run] step={name} elapsed_s={elapsed:.3f} exit={result.returncode} log={log}",
flush=True)
if stamped and (first_stamp(output) != expected_stamp or current_stamp() != expected_stamp):
raise RuntimeError(f"{name} source stamp mismatch; see {log}")
if result.returncode != expected_exit:
raise RuntimeError(f"{name} exit={result.returncode}, expected={expected_exit}; see {log}")
return output
def route_frame(manifest_path: Path) -> tuple[tuple[float, float, float],
tuple[float, float, float]]:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
strata = [entry for entry in manifest["plan"]["strates"]
if entry.get("route") and len(entry["route"]) >= 2]
if not strata:
raise RuntimeError(f"no planned SurfaceWorld route in {manifest_path}")
route = strata[0]["route"]
points = [tuple(float(value) / Q_PER_METRE for value in item["position_q"])
for item in route]
lengths = [math.dist(a, b) for a, b in zip(points, points[1:])]
total = sum(lengths)
if total <= 1.0e-6:
raise RuntimeError(f"degenerate planned route in {manifest_path}")
def at(distance: float) -> tuple[float, float, float]:
remaining = max(0.0, min(total, distance))
for index, (a, b, length) in enumerate(zip(points, points[1:], lengths)):
if remaining <= length or index == len(lengths) - 1:
if length <= 1.0e-9:
return b
t = remaining / length
return tuple(a[axis] + (b[axis] - a[axis]) * t for axis in range(3))
remaining -= length
return points[-1]
center = at(total * 0.5)
before, after = at(total * 0.5 - 1.0), at(total * 0.5 + 1.0)
dx, dy = after[0] - before[0], after[1] - before[1]
length = math.hypot(dx, dy)
if length <= 1.0e-9:
raise RuntimeError(f"degenerate route tangent in {manifest_path}")
return center, (dx / length, dy / length, 0.0)
def camera_angles(camera, target) -> tuple[float, float]:
dx, dy, dz = (target[index] - camera[index] for index in range(3))
return math.degrees(math.atan2(dx, dy)), math.degrees(math.atan2(dz, math.hypot(dx, dy)))
def point_toward(camera, target) -> tuple[float, ...]:
yaw, pitch = camera_angles(camera, target)
return (*camera, yaw, pitch)
def stamp(layout: dict, stable_id: str) -> dict:
for record in layout.get("stamps", []):
if record.get("stable_id") == stable_id:
return record
raise RuntimeError(f"missing resolved stamp {stable_id}")
def make_views(name: str, manifest_path: Path, layout_path: Path):
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
layout = json.loads(layout_path.read_text(encoding="utf-8"))
center, tangent = route_frame(manifest_path)
tx, ty, _ = tangent
nx, ny = -ty, tx
base_height = manifest["recipe"]["surface_profile"]["h0_q"] / Q_PER_METRE
if name == "flatField-monolith":
monolith = stamp(layout, "liminal/oversized-familiar-monolith-v1")
sx, sy, _ = (value / 1.0e6 for value in monolith["position_um"])
floor = monolith["floor_um"] / 1.0e6
eye_camera = (sx - tx * 36.0, sy - ty * 36.0, floor + 1.6)
eye_target = (sx, sy, floor + 10.5)
wide_camera = (sx - tx * 42.0 - nx * 18.0,
sy - ty * 42.0 - ny * 18.0, floor + 42.0)
wide_target = (sx, sy, floor + 8.0)
return {
"eye": (eye_camera, eye_target, 20),
"wide": (wide_camera, wide_target, 24),
"height_center": floor,
"height_range": 38.0,
}
if name == "periodicHills":
# Route points are capsule centers: +0.72 m places the view at 1.60 m above ground.
eye_camera = (center[0], center[1], center[2] + 0.72)
eye_target = (center[0] - tx * 360.0, center[1] - ty * 360.0,
center[2] + 0.72)
wide_camera = (center[0] - tx * 36.0 - nx * 28.0,
center[1] - ty * 36.0 - ny * 28.0, center[2] + 30.0)
wide_target = (center[0] - tx * 310.0, center[1] - ty * 310.0,
center[2] + 0.72)
return {
"eye": (eye_camera, eye_target, 20),
"wide": (wide_camera, wide_target, 28),
"height_center": base_height,
"height_range": 6.0,
}
if name == "road-in-clouds":
deck = stamp(layout,"liminal/road-clouds-v2/deck/000")
path_um = deck.get("sweep_path_um", [])
if len(path_um) < 2:
raise RuntimeError("road layout has no resolved continuous swept path")
points = [tuple(value / 1.0e6 for value in point) for point in path_um]
half_thickness = deck["half_extents_um"][2] / 1.0e6
top = lambda point: point[2] + half_thickness
index = max(1, len(points) // 12)
camera_dx = points[index + 1][0] - points[index - 1][0]
camera_dy = points[index + 1][1] - points[index - 1][1]
camera_tangent_length = math.hypot(camera_dx, camera_dy)
camera_tx, camera_ty = (camera_dx / camera_tangent_length,
camera_dy / camera_tangent_length)
camera_nx, camera_ny = -camera_ty, camera_tx
half_width = deck["half_extents_um"][1] / 1.0e6
eye_target_index = min(len(points) - 1, index + 25)
eye_camera = (points[index][0] + camera_nx * (half_width + 2.0),
points[index][1] + camera_ny * (half_width + 2.0),
top(points[index]) + 1.6)
eye_target = (points[eye_target_index][0] + camera_nx * (half_width - 0.4),
points[eye_target_index][1] + camera_ny * (half_width - 0.4),
top(points[eye_target_index]) + 0.3)
middle = len(points) // 2
dx = points[middle + 1][0] - points[middle - 1][0]
dy = points[middle + 1][1] - points[middle - 1][1]
tangent_length = math.hypot(dx, dy)
tx, ty = dx / tangent_length, dy / tangent_length
nx, ny = -ty, tx
wide_target_index = min(len(points) - 1, middle + 30)
wide_camera = (points[middle][0] - tx * 35.0 + nx * 30.0,
points[middle][1] - ty * 35.0 + ny * 30.0,
top(points[middle]) + 25.0)
wide_target = (points[wide_target_index][0], points[wide_target_index][1],
top(points[wide_target_index]) - 5.0)
cloud_camera = (points[middle][0] - tx * 18.0 + nx * 12.0,
points[middle][1] - ty * 18.0 + ny * 12.0,
base_height + 1.6)
cloud_target = (points[middle][0] + tx * 8.0,
points[middle][1] + ty * 8.0,
points[middle][2] - half_thickness * 0.35)
return {
"eye": (eye_camera, eye_target, 3),
"wide": (wide_camera, wide_target, 3),
"cloud-up": (cloud_camera, cloud_target, 3),
"height_center": base_height + 23.0,
"height_range": 60.0,
"road_deck": top(points[middle]),
"road_width": 2.0 * half_width,
}
if name == "pillar-grid":
pillars = [row for row in layout.get("stamps", [])
if row.get("stable_id", "").startswith("liminal/horizon-pillar-grid-v1/pillar/")]
if len(pillars) < 9:
raise RuntimeError("pillar-grid layout has too few resolved supports")
xs = [row["position_um"][0] / 1.0e6 for row in pillars]
ys = [row["position_um"][1] / 1.0e6 for row in pillars]
floor = min((row["position_um"][2] - row["half_extents_um"][2]) / 1.0e6
for row in pillars)
cx, cy = (min(xs) + max(xs)) * 0.5, (min(ys) + max(ys)) * 0.5
eye_camera = (min(xs) - 78.0, cy, floor + 1.6)
eye_target = (max(xs) + 90.0, cy, floor + 1.6)
wide_camera = (cx - 280.0, cy - 420.0, floor + 230.0)
wide_target = (cx, cy, floor + 6.0)
return {
"eye": (eye_camera, eye_target, 3),
"wide": (wide_camera, wide_target, 3),
"height_center": floor + 12.0,
"height_range": 42.0,
}
raise RuntimeError(f"no capture framing for fixture {name}")
def headless_command(binary: Path, fixture: Path, camera, mesh: Path,
manifest: Path, layout: Path, workers: int,
clip_radius: int, *, hash_tiles: bool = False,
audit_road: bool = False, max_level: int = 0,
far_lods: bool = False,
max_clip_override: int | None = None) -> list[object]:
max_clip_level = (max_clip_override if max_clip_override is not None else
8 if far_lods else max_level)
command: list[object] = [binary, fixture, f"--workers={workers}",
"--seed=0", "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
"--force-generate", f"--clip-radius={clip_radius}",
f"--max-clip-level={max_clip_level}",
f"--max-level={max_level}", "--render-distance-chunks=0",
f"--camera={fmt(point_toward(camera, (camera[0], camera[1] + 1.0, camera[2])))}",
f"--mesh-export={mesh}", f"--world-assembler-manifest-output={manifest}",
f"--resolved-layout-output={layout}"]
if hash_tiles:
command.append("--stream-hash-tiles")
if audit_road:
command.append("--audit-liminal-road-walk")
return command
def determinism_digest(output: str, manifest_path: Path,
layout_path: Path) -> tuple[str, str, str]:
tile_hashes = sorted(line for line in output.splitlines()
if line.startswith("[PipelineTileHash] "))
if not manifest_path.is_file() or not layout_path.is_file() or not tile_hashes:
raise RuntimeError("determinism run omitted manifest, layout or per-tile hashes")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
layout = json.loads(layout_path.read_text(encoding="utf-8"))
manifest_digest = (manifest["content_hash"],manifest["recipe"]["content_hash"],
manifest["plan"]["content_hash"])
return ("|".join(manifest_digest),layout["stable_identity"],"\n".join(tile_hashes))
def run_determinism(name: str, fixture: Path, view, *, binaries: dict[str, Path],
out: Path, env: dict[str, str], max_level: int = 0,
far_lods: bool = False, max_clip_override: int | None = None) -> None:
camera, _, _ = view["eye"]
clip_radius = 1
max_clip = 0
baseline = None
cells = 0
for compiler, workers in ((compiler, workers)
for compiler in ("clang", "gcc")
for workers in (1, 8, 12)):
tag = f"{name}-{compiler}-w{workers}"
run_dir = out / "determinism" / tag
run_dir.mkdir(parents=True, exist_ok=True)
mesh = run_dir / "feature.vfmesh"
manifest = run_dir / "manifest.json"
layout = run_dir / "layout.json"
command = headless_command(binaries[compiler], fixture,camera,mesh,manifest,
layout,workers,max_clip,hash_tiles=True,max_level=max_level,
far_lods=far_lods,max_clip_override=max_clip_override)
output = run_step(tag,command,out/"logs"/f"determinism-{tag}.log",
env=env,stamped=True)
digest = determinism_digest(output,manifest,layout)
if baseline is None:
baseline = digest
elif digest != baseline:
raise RuntimeError(f"{name} diverged at {compiler} W{workers}; see determinism logs")
cells += 1
print(f"[L143Determinism] fixture={name} compiler_worker_cells={cells} identical=yes status=PASS",
flush=True)
def export_and_capture(name: str, fixture: Path, views: dict, *,
headless: Path, renderer: Path, out: Path,
artifact_dir: Path, env: dict[str, str]) -> None:
view_names = ("eye", "wide", "cloud-up") if name == "road-in-clouds" else ("eye", "wide")
for view_name in view_names:
camera, target, clip_radius = views[view_name]
clip_radius = min(clip_radius,3)
stem = f"{name}-{view_name}"
mesh = out / "capture-meshes" / f"{stem}.vfmesh"
manifest = out / "capture-manifests" / f"{stem}.json"
layout = out / "capture-layouts" / f"{stem}.json"
png = out / "capture-png" / f"{stem}.png"
image = artifact_dir / f"{stem}.jpg"
for path in (mesh,manifest,layout,png,image):
path.parent.mkdir(parents=True,exist_ok=True)
command = headless_command(headless,fixture,camera,mesh,manifest,layout,12,
clip_radius,max_level=-1,far_lods=True,
max_clip_override=2 if name == "road-in-clouds" else None)
output = run_step(f"capture-{stem}-export",command,
out/"logs"/f"capture-{stem}-export.log",env=env,stamped=True)
if not mesh.is_file() or mesh.stat().st_size == 0:
raise RuntimeError(f"{stem} produced an empty mesh")
yaw, pitch = camera_angles(camera,target)
height_center = views["height_center"]
height_range = views["height_range"]
if name == "road-in-clouds":
fog_start,fog_end,fog_color=(24,260,"0.88,0.93,0.95")
elif name == "pillar-grid":
fog_start,fog_end,fog_color=220,1250,"0.82,0.88,0.91"
elif name == "periodicHills":
fog_start,fog_end,fog_color=210,1000,"0.78,0.84,0.88"
else:
fog_start,fog_end,fog_color=30,105,FOG_COLOR
renderer_command = [renderer,mesh,png,"--size=1600x900","--fov-deg=72",
f"--camera-m={fmt(camera)}",f"--yaw-deg={yaw:.6f}",
f"--pitch-deg={pitch:.6f}","--tile-skirts=on","--shade=readable",
"--headlamp=off","--height-tint=on",f"--height-center-m={height_center:.6f}",
f"--height-range-m={height_range:.6f}","--light-direction=-0.42,-0.58,0.70",
"--fog=on",f"--fog-start-m={fog_start}",f"--fog-end-m={fog_end}",
f"--fog-color={fog_color}"]
run_step(f"capture-{stem}-render",renderer_command,
out/"logs"/f"capture-{stem}-render.log",env=env,stamped=True)
run_step(f"capture-{stem}-jpeg",["ffmpeg","-nostdin","-hide_banner","-loglevel",
"error","-y","-i",png,"-frames:v","1","-q:v","4",image],
out/"logs"/f"capture-{stem}-jpeg.log",env=env)
quality = run_step(f"capture-{stem}-quality",
[sys.executable,ROOT/"Standalone/Tools/verify/capture_quality.py",image],
out/"logs"/f"capture-{stem}-quality.log",env=env)
if "[CaptureQuality]" not in quality or not image.is_file() \
or image.stat().st_size > 300_000:
raise RuntimeError(f"{stem} capture failed quality or size checks")
eye_height = "1.600m" if view_name in ("eye", "cloud-up") else "wide"
print(f"[L143Capture] fixture={name} view={view_name} eye_height={eye_height} "
f"camera_m=({fmt(camera)}) target_m=({fmt(target)}) bytes={image.stat().st_size} "
f"sha256={sha256(image)} status=PASS",flush=True)
def run_road_walk_audit(fixture: Path, view, *, headless: Path,
out: Path, env: dict[str, str]) -> None:
camera = view["eye"][0]
run_dir = out / "road-walk-audit"
run_dir.mkdir(parents=True,exist_ok=True)
output = run_step("road-walk-audit",headless_command(headless,fixture,camera,
run_dir/"audit.vfmesh",run_dir/"manifest.json",run_dir/"layout.json",12,1,
audit_road=True),out/"logs"/"road-walk-audit.log",env=env,stamped=True)
if not re.search(r"\[LiminalRoadWalkAudit\].*status=PASS",output):
raise RuntimeError("raised road failed the Core capsule walk audit")
def run_road_short_negative_control(fixture: Path, view, *, headless: Path,
out: Path, env: dict[str, str]) -> None:
camera = view["eye"][0]
run_dir = out / "road-short-negative-control"
run_dir.mkdir(parents=True,exist_ok=True)
output = run_step("road-short-negative-control",
headless_command(headless,fixture,camera,run_dir/"control.vfmesh",
run_dir/"manifest.json",run_dir/"layout.json",12,1,audit_road=True),
out/"logs"/"road-short-negative-control.log",env=env,stamped=True,
expected_exit=14)
match = re.search(r"\[LiminalRoadWalkAudit\].*path_length_m=([0-9.]+).*status=FAIL",output)
if not match or float(match.group(1)) >= 300.0:
raise RuntimeError("short-road negative control did not falsify the length gate")
print(f"[L143RoadNegativeControl] path_length_m={match.group(1)} "
"expected=FAIL status=PASS",flush=True)
def retain_unchanged_captures(artifact_dir: Path) -> None:
reference_dir = Path(__file__).resolve().parents[1] / "Linux142" / "captures"
for name in ("flatField-monolith", "periodicHills"):
for view in ("eye", "wide"):
filename = f"{name}-{view}.jpg"
source, destination = reference_dir / filename, artifact_dir / filename
if not source.is_file():
raise RuntimeError(f"missing L142 reference capture {source}")
shutil.copy2(source,destination)
if sha256(source) != sha256(destination):
raise RuntimeError(f"unchanged reference capture copy differs: {filename}")
print(f"[L143Capture] fixture={name} view={view} retained_from=Linux142 "
f"sha256={sha256(destination)} status=PASS",flush=True)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--out", type=Path, default=OUT_DEFAULT)
parser.add_argument("--artifact-dir", type=Path, default=CAPTURE_DEFAULT)
parser.add_argument("--headless-clang", type=Path, required=True)
parser.add_argument("--headless-gcc", type=Path, required=True)
parser.add_argument("--renderer", type=Path, required=True)
args = parser.parse_args()
out = args.out.resolve()
artifact_dir = args.artifact_dir.resolve()
fixture_dir = out / "fixtures"
for directory in (out,fixture_dir,artifact_dir):
directory.mkdir(parents=True,exist_ok=True)
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")
run_step("build-fixtures",[sys.executable,
ROOT/"Standalone/Research/WorldAssemblerM3/Linux143/BuildL143Fixtures.py",
f"--out={fixture_dir}"],out/"logs"/"build-fixtures.log",env=env)
binaries = {"clang":args.headless_clang.resolve(),"gcc":args.headless_gcc.resolve()}
fixtures = {
"flatField-monolith":fixture_dir/"liminal-flatField-monolith.json",
"periodicHills":fixture_dir/"liminal-periodicHills.json",
"road-in-clouds":fixture_dir/"liminal-road-in-clouds.json",
"pillar-grid":fixture_dir/"liminal-pillar-grid.json",
}
probed = {}
for name,fixture in fixtures.items():
probe_dir=out/"plan-probes"/name
probe_dir.mkdir(parents=True,exist_ok=True)
camera=(0.0,0.0,-500.0)
output=run_step(f"plan-probe-{name}",headless_command(binaries["clang"],fixture,
camera,probe_dir/"probe.vfmesh",probe_dir/"manifest.json",
probe_dir/"layout.json",12,1),out/"logs"/f"plan-probe-{name}.log",
env=env,stamped=True)
if "[WorldAssemblerFullManifest]" not in output:
raise RuntimeError(f"{name} probe omitted its full plan manifest")
probed[name]=make_views(name,probe_dir/"manifest.json",probe_dir/"layout.json")
for name,fixture in fixtures.items():
run_determinism(name,fixture,probed[name],binaries=binaries,out=out,env=env)
run_determinism("pillar-grid-far-lod",fixtures["pillar-grid"],probed["pillar-grid"],
binaries=binaries,out=out,env=env,max_level=-1,far_lods=True,max_clip_override=8)
run_road_walk_audit(fixtures["road-in-clouds"],probed["road-in-clouds"],
headless=binaries["clang"],out=out,env=env)
run_road_short_negative_control(fixture_dir/"liminal-road-short-control.json",
probed["road-in-clouds"],headless=binaries["clang"],out=out,env=env)
retain_unchanged_captures(artifact_dir)
for name in ("road-in-clouds", "pillar-grid"):
fixture=fixtures[name]
export_and_capture(name,fixture,probed[name],headless=binaries["clang"],
renderer=args.renderer.resolve(),out=out,artifact_dir=artifact_dir,env=env)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except Exception as error:
print(f"[L143Capture] status=FAIL error={error}",file=sys.stderr,flush=True)
raise SystemExit(1)