#!/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)