#!/usr/bin/env python3 """Build isolated L172 UE trees, measure admission, and rescore saved traces.""" from __future__ import annotations import argparse import datetime as dt import hashlib import json import math import os from pathlib import Path import re import shutil import statistics import subprocess import time from typing import Any ROOT = Path(__file__).resolve().parents[4] HOST = Path("/home/jahni/VFHost") ENGINE = Path("/home/jahni/UnrealEngine") MAIN_CHECKOUT = Path("/home/jahni/VoxelM/Plugins/VoxelForge") OUT = ROOT / "Saved/Linux/l172" UE_OUT = OUT / "ue" BUILD_OUT = OUT / "builds" BIN_OUT = OUT / "bin" HOST_PLUGIN_LINK = HOST / "PluginDeps/VoxelForge" HOST_PLUGIN_BINARY = HOST / "Binaries/Linux/libUnrealEditor-VoxelForge.so" EDITOR = ENGINE / "Engine/Binaries/Linux/UnrealEditor-Cmd" BUILD_SH = ENGINE / "Engine/Build/BatchFiles/Linux/Build.sh" DWARFDUMP = Path("/usr/lib/llvm-19/bin/llvm-dwarfdump") INTERMEDIATE = HOST / "Intermediate" MOVE_START_S = 6.0 MOVE_DURATION_S = 12.0 SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375) LINK_NAMES = ("Source", "Standalone", "VoxelForge.uplugin") OBJECTS = { "voxel_world": "External/Build/Linux/x64/UnrealEditor/Development/VoxelForge/VoxelWorld.cpp.o", "core_adapter": "External/Build/Linux/x64/UnrealEditor/Development/VoxelForge/VoxelForgeCoreAdapter.cpp.o", "core_api_shim": "External/Build/Linux/x64/UnrealEditor/Development/VoxelForgeCore/VFCore_Core_src_VoxelForgeCore_cpp.cpp.o", } WAIT_OFF = 1 APPLY_OFF_MS = 0.0 WAIT_ON = 0 APPLY_ON_MS = 1.0 def utc_now() -> str: return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") def write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") temporary.replace(path) def git(root: Path, *args: str) -> str: result = subprocess.run(["git", "-C", str(root), *args], text=True, capture_output=True, check=False) if result.returncode != 0: raise RuntimeError(f"git -C {root} {' '.join(args)} failed: {result.stderr.strip()}") return result.stdout.rstrip("\n") def root_info(root: Path) -> dict[str, Any]: actual_root = Path(git(root, "rev-parse", "--show-toplevel")).resolve() expected_root = root.resolve() if actual_root != expected_root: raise RuntimeError(f"plugin root mismatch: requested={expected_root}, git={actual_root}") return { "plugin_root": str(actual_root), "git_head": git(actual_root, "rev-parse", "HEAD"), "git_status_porcelain": git(actual_root, "status", "--porcelain"), } def link_roots() -> dict[str, str]: roots = {} for name in LINK_NAMES: link = HOST_PLUGIN_LINK / name if not link.is_symlink(): raise RuntimeError(f"VFHost plugin dependency is not a symlink: {link}") resolved = link.resolve(strict=True) roots[name] = str(resolved.parent) return roots def assert_shared_root(expected_root: Path) -> dict[str, str]: expected = expected_root.resolve(strict=True) roots = link_roots() if any(Path(root) != expected for root in roots.values()): raise RuntimeError(f"VFHost plugin symlinks do not share expected root {expected}: {roots}") return roots def _move_plugin_intermediates(old_root: str | None, new_root: Path) -> list[dict[str, str]]: matches = [] if INTERMEDIATE.exists(): for current, dirs, _files in os.walk(INTERMEDIATE, topdown=True): for dirname in list(dirs): if dirname.startswith("VoxelForge"): matches.append(Path(current) / dirname) dirs.remove(dirname) if not matches: return [] head = "unknown" if old_root: try: head = root_info(Path(old_root))["git_head"][:12] except Exception: head = Path(old_root).name archive = OUT / "intermediate-archive" / f"{time.time_ns()}-{head}-to-{new_root.name}" moved = [] for source in sorted(matches, key=lambda path: len(path.parts)): relative = source.relative_to(INTERMEDIATE) target = archive / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(source), str(target)) moved.append({"from": str(source), "to": str(target)}) return moved def switch_plugin_root(target_root: Path) -> dict[str, Any]: target = target_root.resolve(strict=True) if not (target / "Source").is_dir() or not (target / "Standalone").is_dir(): raise RuntimeError(f"tree does not contain its own Source and Standalone: {target}") if not (target / "VoxelForge.uplugin").is_file(): raise RuntimeError(f"tree has no plugin descriptor: {target}") current_roots = set(link_roots().values()) old_root = next(iter(current_roots)) if len(current_roots) == 1 else None moved = [] if old_root != str(target): moved = _move_plugin_intermediates(old_root, target) for name in LINK_NAMES: link = HOST_PLUGIN_LINK / name temporary = link.with_name(f"{link.name}.l172-{os.getpid()}.tmp") if temporary.exists() or temporary.is_symlink(): temporary.unlink() temporary.symlink_to(target / name) os.replace(temporary, link) roots = assert_shared_root(target) return {"active_root": str(target), "links": roots, "intermediates_moved": moved} def restore_main_checkout() -> dict[str, Any]: return switch_plugin_root(MAIN_CHECKOUT) def sha256_file(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 dwarf_struct_size(path: Path) -> int: if not DWARFDUMP.is_file(): raise RuntimeError(f"llvm-dwarfdump is missing: {DWARFDUMP}") result = subprocess.run([str(DWARFDUMP), "--name=WorldAssemblyStratumPlanV1", str(path)], text=True, capture_output=True, check=False) if result.returncode != 0: raise RuntimeError(f"llvm-dwarfdump failed for {path}: {result.stderr.strip()}") dies = re.split(r"(?=0x[0-9a-fA-F]+: DW_TAG_)", result.stdout) matches = [] for die in dies: if not re.search(r'DW_AT_name\s+\("WorldAssemblyStratumPlanV1"\)', die): continue size = re.search(r"DW_AT_byte_size\s+\(0x([0-9a-fA-F]+)\)", die) if size: matches.append(int(size.group(1), 16)) if len(matches) != 1: raise RuntimeError(f"expected one complete plan type in {path}; sizes={matches}") return matches[0] def dwarf_pre_run_gate() -> dict[str, Any]: sizes = {} for label, relative in OBJECTS.items(): path = INTERMEDIATE / relative if not path.is_file(): raise RuntimeError(f"pre-run DWARF gate object is missing: {path}") sizes[label] = {"object": str(path), "byte_size": dwarf_struct_size(path)} unique = {item["byte_size"] for item in sizes.values()} if len(unique) != 1: raise RuntimeError(f"WorldAssemblyStratumPlanV1 size mismatch: {sizes}") return {"status": "PASS", "struct": "WorldAssemblyStratumPlanV1", "objects": sizes} def create_fresh_baseline_worktree() -> Path: timestamp = str(time.time_ns()) repo_host = OUT / "worktree-hosts" / f"ca0bf76-git-{timestamp}" parent = OUT / "worktrees" parent.mkdir(parents=True, exist_ok=True) repo_host.parent.mkdir(parents=True, exist_ok=True) commit = git(ROOT, "rev-parse", "ca0bf76") clone = subprocess.run(["git", "clone", "--shared", "--no-checkout", str(MAIN_CHECKOUT), str(repo_host)], text=True, capture_output=True, check=False) if clone.returncode != 0: raise RuntimeError(f"isolated git worktree host clone failed: {clone.stderr.strip()}") fetch = subprocess.run(["git", "-C", str(repo_host), "fetch", "--no-tags", str(MAIN_CHECKOUT), commit], text=True, capture_output=True, check=False) if fetch.returncode != 0: raise RuntimeError(f"ca0bf76 fetch into isolated worktree host failed: " f"{fetch.stderr.strip()}") path = parent / f"assembler-world-ca0bf76-fresh-{timestamp}" result = subprocess.run(["git", "-C", str(repo_host), "worktree", "add", "--detach", str(path), commit], text=True, capture_output=True, check=False) if result.returncode != 0: raise RuntimeError(f"fresh ca0bf76 worktree add failed: {result.stderr.strip()}") info = root_info(path) if not info["git_head"].startswith("ca0bf767ca7afe40723bf8bb6987d9efc9ea7523"): raise RuntimeError(f"fresh baseline worktree has unexpected HEAD: {info}") return path.resolve() def build_tree(root: Path, label: str) -> dict[str, Any]: switch = switch_plugin_root(root) source = root_info(root) if len(set(switch["links"].values())) != 1: raise RuntimeError(f"link roots diverged before build: {switch}") log = BUILD_OUT / f"ubt-{label}.log" log.parent.mkdir(parents=True, exist_ok=True) command = [str(BUILD_SH), "VFHostEditor", "Linux", "Development", f"-Project={HOST / 'VFHost.uproject'}", "-NoUBA", "-MaxParallelActions=4", f"-Log={log}"] started = time.monotonic() with log.open("w", encoding="utf-8") as stream: result = subprocess.run(["taskset", "-c", "0-11", "timeout", "1800s", *command], cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1800) elapsed = time.monotonic() - started output = log.read_text(encoding="utf-8", errors="replace") if result.returncode != 0: raise RuntimeError(f"UE build {label} failed ({result.returncode}); log={log}") if not re.search(r"Compile VoxelWorld\.cpp", output): raise RuntimeError(f"UE build {label} did not compile VoxelWorld.cpp; log={log}") gate = dwarf_pre_run_gate() if not HOST_PLUGIN_BINARY.is_file(): raise RuntimeError(f"UE build did not produce {HOST_PLUGIN_BINARY}") artifact_dir = BIN_OUT / label artifact_dir.mkdir(parents=True, exist_ok=True) artifact = artifact_dir / "libUnrealEditor-VoxelForge.so" shutil.copy2(HOST_PLUGIN_BINARY, artifact) return { **source, "build_command": command, "build_log": str(log), "compiled_voxel_world_cpp": True, "elapsed_s": round(elapsed, 3), "dwarf_pre_run_gate": gate, "binary": str(artifact), "binary_sha256": sha256_file(artifact), "active_links": switch["links"], "intermediates_moved_before_build": switch["intermediates_moved"], } def prepare() -> int: OUT.mkdir(parents=True, exist_ok=True) baseline_root = None builds = {} try: baseline_root = create_fresh_baseline_worktree() builds["ca0bf76"] = build_tree(baseline_root, "ca0bf76") builds["ca0bf76"]["worktree_root"] = str(baseline_root) builds["ca0bf76"]["worktree_created_with"] = [ "git", "worktree", "add", "--detach", str(baseline_root), "ca0bf76"] write_json(OUT / "builds.json", {"status": "BASELINE_READY", **builds}) builds["fixed"] = build_tree(ROOT, "fixed") builds["fixed"]["worktree_root"] = str(ROOT.resolve()) write_json(OUT / "builds.json", {"status": "BOTH_TREES_READY", **builds}) print("[L172Builds] " + json.dumps({ key: {"head": value["git_head"], "root": value["plugin_root"], "binary_sha256": value["binary_sha256"], "dwarf_sizes": {name: item["byte_size"] for name, item in value["dwarf_pre_run_gate"]["objects"].items()}} for key, value in builds.items() if key != "status"}, sort_keys=True), flush=True) return 0 finally: restored = restore_main_checkout() write_json(OUT / "restored-main-links.json", { "time_utc": utc_now(), "main_root": str(MAIN_CHECKOUT.resolve()), "links": restored["links"], "intermediates_moved": restored["intermediates_moved"]}) def read_events(path: Path) -> list[dict[str, Any]]: rows = [] if not path.is_file(): return rows for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.startswith("{"): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def first_frame_arrival(events: list[dict[str, Any]]) -> dict[str, Any] | None: for event in events: if event.get("type") != "scheduler_admission": continue missing = event.get("desired_missing_by_lod", []) pending = event.get("pending_by_lod", []) if (len(missing) >= 3 and len(pending) >= 3 and sum(int(value) for value in missing[:3]) == 0 and sum(int(value) for value in pending[:3]) == 0): return event return None def percentile(values: list[float], fraction: float = 0.99) -> float | None: if not values: return None ordered = sorted(values) index = max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1)) return ordered[index] def percent(numerator: int | float, denominator: int | float) -> float | None: return 100.0 * numerator / denominator if denominator > 0 else None def summarize_trial(condition: str, trace: Path, result: subprocess.CompletedProcess[str], wall_s: float, expected_wait: int, expected_budget_ms: float) -> dict[str, Any]: events = read_events(trace) backend = next((event for event in events if event.get("type") == "worker_backend_config"), {}) scheduler = [event for event in events if event.get("type") == "scheduler_admission"] if not scheduler: raise RuntimeError(f"no scheduler_admission readback in {trace}") if (int(backend.get("streaming_fast_travel_wait_for_all_critical_tiles", -1)) != expected_wait or abs(float(backend.get("streaming_fast_travel_apply_budget_ms", -999.0)) - expected_budget_ms) > 0.01 or int(backend.get("effective_tile_slots", -1)) != 8): raise RuntimeError(f"switch or worker readback mismatch in {trace}: {backend}") if any(int(event.get("forward_first", -1)) != 0 or int(event.get("rebuild_on_core_region_completion", -1)) != 1 or int(event.get("core_desired_cache_by_center", -1)) != 0 or int(event.get("incremental_desired_set", -1)) != 1 or int(event.get("incremental_desired_audit_every", -1)) != 0 or int(event.get("incremental_desired_materialization_audit_every", -1)) != 0 or int(event.get("incremental_desired_audit_inject_mismatch", -1)) != 0 for event in scheduler): raise RuntimeError(f"non-default scheduler CVar readback in {trace}") arrival = first_frame_arrival(scheduler) if arrival is None: raise RuntimeError(f"missing frame-resolution LOD0-2 settled event in {trace}") if condition == "fixed-off": arrival_s = float(arrival["t_s"]) pre_arrival = [event for event in scheduler if float(event.get("t_s", 0.0)) < arrival_s] post_arrival = [event for event in scheduler if float(event.get("t_s", 0.0)) >= arrival_s] if any(int(event.get("fast_travel_mode", 0)) != 0 for event in pre_arrival): raise RuntimeError(f"fixed-off entered fast-travel mode before initial arrival: {trace}") if not any(int(event.get("fast_travel_mode", 0)) == 1 for event in post_arrival): raise RuntimeError(f"fixed-off never latched fast-travel mode after initial arrival: {trace}") if condition == "fixed-on": arrival_s = float(arrival["t_s"]) pre_arrival = [event for event in scheduler if float(event.get("t_s", 0.0)) < arrival_s] apply_events = [event for event in events if event.get("type") == "apply_frame"] if any(int(event.get("fast_travel_mode", 0)) != 0 for event in pre_arrival): raise RuntimeError(f"fast-travel mode activated before frame-settled arrival: {trace}") if any(float(event.get("apply_budget_ms", 0.0)) < 1.99 and float(event.get("t_s", 0.0)) <= arrival_s for event in apply_events): raise RuntimeError(f"fast-travel apply budget activated before arrival: {trace}") if not any(int(event.get("initial_arrival_set_settled", 0)) == 1 and int(event.get("fast_travel_mode", 0)) == 1 for event in scheduler): raise RuntimeError(f"fixed-on latch did not activate after settled arrival: {trace}") if not any(float(event.get("apply_budget_ms", 0.0)) <= 1.001 for event in apply_events): raise RuntimeError(f"fixed-on 1 ms apply budget never activated: {trace}") move_stop = next((event for event in events if event.get("type") == "streaming_test_move_stopped"), None) if move_stop is None: raise RuntimeError(f"missing move-stop event in {trace}") stop_s = float(move_stop.get("t_s", 0.0)) samples = [event for event in events if event.get("type") == "streaming_sample"] moving = [event for event in samples if MOVE_START_S <= float(event.get("t_s", 0.0)) < stop_s and float(event.get("speed_cm_s", 0.0)) >= 50.0] dwell = [event for event in samples if float(event.get("t_s", 0.0)) >= stop_s and float(event.get("speed_cm_s", 0.0)) < 50.0] if result.returncode != 0 or len(moving) < 40 or len(dwell) < 16: raise RuntimeError(f"incomplete run {trace}: exit={result.returncode}, " f"moving={len(moving)}, dwell={len(dwell)}") work_start = min(float(event.get("t_s", 0.0)) for event in moving) frames = [event for event in events if event.get("type") == "game_thread_work_frame" and work_start <= float(event.get("t_s", 0.0)) < stop_s] if not frames: raise RuntimeError(f"missing moving game-thread frames in {trace}") near, far_holes = [], [] for event in moving: desired = event.get("desired_by_lod", [0] * 9) resident = event.get("resident_by_lod", [0] * 9) covered = event.get("covered_by_lod", resident) near.append(percent(sum(resident[:3]), sum(desired[:3])) or 0.0) far_holes.append(percent(sum(desired[3:]) - sum(covered[3:]), sum(desired[3:])) or 0.0) settled_sample = next((event for event in samples if float(event.get("t_s", 0.0)) < work_start and float(event.get("speed_cm_s", 0.0)) < 50.0 and sum(event.get("settled_by_lod", [0] * 9)[:3]) >= sum(event.get("desired_by_lod", [0] * 9)[:3]) > 0), None) return { "condition": condition, "moving_samples": len(moving), "dwell_samples": len(dwell), "effective_tile_slots": backend.get("effective_tile_slots"), "frame_resolution_arrival_s": float(arrival["t_s"]), "frame_resolution_arrival_frame": int(arrival.get("frame", -1)), "streaming_sample_arrival_s": (float(settled_sample["t_s"]) if settled_sample else None), "near_readiness_moving_mean_pct": statistics.mean(near), "far_holes_moving_median_pct": statistics.median(far_holes), "tick_p99_ms": percentile([float(frame.get("tick_ms", 0.0)) for frame in frames]), "tick_worst_ms": max(float(frame.get("tick_ms", 0.0)) for frame in frames), "early_admission_wait_for_critical": expected_wait, "early_admission_apply_budget_ms": expected_budget_ms, "trace": str(trace.relative_to(ROOT)), "wall_s": round(wall_s, 3), } def run_trial(name: str, batch: str, condition: str, root: Path, binary: Path, wait: int, apply_ms: float, include_switch_overrides: bool = True) -> dict[str, Any]: switch = switch_plugin_root(root) identity = root_info(root) assert_shared_root(root) if not binary.is_file(): raise FileNotFoundError(binary) binary_hash = sha256_file(binary) shutil.copy2(binary, HOST_PLUGIN_BINARY) run_dir = UE_OUT / batch / condition run_dir.mkdir(parents=True, exist_ok=True) export = HOST / "Plugins/VoxelForge/Saved/Linux/l172/export" / batch / condition / name userdir = HOST / "Saved/Linux/l172/user" / batch / condition / name zendata = HOST / "Saved/Linux/l172/zen" / batch / condition / name ddc = HOST / "Saved/Linux/l172/DDC" for directory in (export, userdir, zendata, ddc): directory.mkdir(parents=True, exist_ok=True) trace = run_dir / f"{name}.trace.jsonl" log = run_dir / f"{name}.log" stdout = run_dir / f"{name}.stdout.log" manifest_path = run_dir / f"{name}.manifest.json" trace.unlink(missing_ok=True) spawn_cm = ",".join(f"{value * 100.0:.3f}" for value in SPAWN_M) command = [ str(EDITOR), str(HOST / "VFHost.uproject"), "-nullrhi", "-unattended", "-nop4", "-nosplash", "-NoSound", "-corelimit=12", "-NoZenAutoLaunch", "-notraceserver", "-nocrashreports", "-DisablePlugins=AndroidFileServer", "-ddc=NoZenLocalFallback", f"-userdir={userdir}", f"-ZenDataPath={zendata}", f"-LocalDataCachePath={ddc}", "-run=VoxelForgeExplore", "-seed=0", "-archetype=TunnelNetwork", "-slot=0", "-modes=export", "-gameconfig=1", "-coreworldtest=1", "-ownerfourstrateplan=1", "-corevoxelsizecm=25", "-coreislandtest=0", "-corefallbacktest=0", "-coreofftest=0", "-coreplantest=0", "-settingsref=/Game/VoxelForge/DA_Settings.DA_Settings", "-corehybridmesher=0", f"-out={export}", "-voxel.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=3000.0", f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}", f"-voxel.TestMoveDurationSeconds={MOVE_DURATION_S:.1f}", "-voxel.TestExitSeconds=0.0", "-voxel.TestMoveDirection=1,0,0", "-voxel.TestExitOnSteady=0", "-coreworldstreamtrace=1", f"-voxel.StartupTraceSpawnCm={spawn_cm}", "-voxel.StartupTraceUntilEndPlay=1", "-voxel.StartupTraceThroughCrossing=1", "-voxel.StartupTraceMaxTiles=16384", f"-voxel.StartupTraceFile={trace}", f"-abslog={log}", ] if include_switch_overrides: command.extend(( f"-voxel.StreamingFastTravelWaitForAllCriticalTiles={wait}", f"-voxel.StreamingFastTravelApplyBudgetMilliseconds={apply_ms:.1f}", )) manifest = { **identity, "created_utc": utc_now(), "batch": batch, "condition": condition, "run": name, "active_links": switch["links"], "intermediates_moved_on_tree_switch": switch["intermediates_moved"], "binary": str(binary), "binary_sha256": binary_hash, "staged_host_binary_sha256": sha256_file(HOST_PLUGIN_BINARY), "expected_switches": {"wait_for_all_critical_tiles": wait, "apply_budget_ms": apply_ms}, "switch_source": ("command_line_overrides" if include_switch_overrides else "compiled_defaults"), "command": command, "trace": str(trace), "unreal_log": str(log), } write_json(manifest_path, manifest) started = time.monotonic() with stdout.open("w", encoding="utf-8") as stream: result = subprocess.run(["taskset", "-c", "0-11", "timeout", "1740s", *command], cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740) wall_s = time.monotonic() - started try: row = summarize_trial(condition, trace, result, wall_s, wait, apply_ms) row["run"] = name manifest["result"] = {"exit_code": result.returncode, "wall_s": round(wall_s, 3), "status": "PASS", "owner_world_ran": True} row["manifest"] = str(manifest_path.relative_to(ROOT)) write_json(manifest_path, manifest) return row except Exception as error: manifest["result"] = {"exit_code": result.returncode, "wall_s": round(wall_s, 3), "status": "FAIL", "error": str(error)} write_json(manifest_path, manifest) raise def frame_arrival_from_trace(path: Path) -> float | None: event = first_frame_arrival(read_events(path)) return float(event["t_s"]) if event is not None else None def rescore_saved_traces() -> dict[str, Any]: result: dict[str, Any] = {"metric": "first scheduler_admission with " "sum(desired_missing_by_lod[0:3]) == 0 and " "sum(pending_by_lod[0:3]) == 0; t_s at that scheduler frame"} for round_no in ("169", "170"): root = ROOT / f"Saved/Linux/l{round_no}/ue/movement" arms = {} for arm in ("baseline", "candidate"): values = {} for trace in sorted((root / arm).glob("pair-*.trace.jsonl")): arrival = frame_arrival_from_trace(trace) if arrival is not None: number = int(re.search(r"pair-(\d+)", trace.name).group(1)) values[number] = {"arrival_s": arrival, "trace": str(trace.relative_to(ROOT))} arms[arm] = values pairs = [] for number in sorted(arms["baseline"].keys() & arms["candidate"].keys()): delta = arms["candidate"][number]["arrival_s"] - arms["baseline"][number]["arrival_s"] pairs.append({"pair": number, "delta_s_candidate_minus_baseline": delta, "candidate_slower": delta > 0, "baseline": arms["baseline"][number], "candidate": arms["candidate"][number]}) deltas = [pair["delta_s_candidate_minus_baseline"] for pair in pairs] result[round_no] = { "pairs": pairs, "baseline_median_arrival_s": statistics.median( [value["arrival_s"] for value in arms["baseline"].values()]) if arms["baseline"] else None, "candidate_median_arrival_s": statistics.median( [value["arrival_s"] for value in arms["candidate"].values()]) if arms["candidate"] else None, "paired_median_delta_ms": 1000.0 * statistics.median(deltas) if deltas else None, "candidate_slower_pairs": sum(delta > 0 for delta in deltas), "pair_count": len(pairs), } path = OUT / "arrival-rescore-l169-l170.json" write_json(path, result) print("[L172ArrivalRescore] " + json.dumps({ key: {metric: value for metric, value in data.items() if metric != "pairs"} for key, data in result.items() if key in ("169", "170")}, sort_keys=True), flush=True) return result def percentile_nearest_rank(values: list[float], fraction: float) -> float: ordered = sorted(values) if not ordered: raise ValueError("percentile requires values") return ordered[max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1))] def paired_deltas(rows: list[dict[str, Any]], first: str, second: str, metric: str) -> list[dict[str, Any]]: by_pair: dict[int, dict[str, dict[str, Any]]] = {} for row in rows: pair_match = re.search(r"pair-(\d+)", row["run"]) if not pair_match: continue pair = int(pair_match.group(1)) by_pair.setdefault(pair, {})[row["condition"]] = row result = [] for pair, arms in sorted(by_pair.items()): if first not in arms or second not in arms: continue a, b = arms[first][metric], arms[second][metric] result.append({"pair": pair, "first": first, "second": second, "delta_s_second_minus_first": b - a, "first_value": a, "second_value": b}) return result def run_batch_arm(rows: list[dict[str, Any]], batch: str, condition: str, root: Path, binary: Path, wait: int, apply_ms: float, run_name: str) -> dict[str, Any]: row = run_trial(run_name, batch, condition, root, binary, wait, apply_ms) rows.append(row) write_json(UE_OUT / batch / "runs-partial.json", rows) return row def load_partial_rows(batch: str) -> list[dict[str, Any]]: path = UE_OUT / batch / "runs-partial.json" if not path.is_file(): return [] rows = json.loads(path.read_text(encoding="utf-8")) if not isinstance(rows, list): raise RuntimeError(f"invalid partial rows file: {path}") return rows def has_run(rows: list[dict[str, Any]], condition: str, run_name: str) -> bool: return any(row.get("condition") == condition and row.get("run") == run_name for row in rows) def require_quiet_machine() -> dict[str, Any]: uptime = subprocess.run(["uptime"], text=True, capture_output=True, check=False).stdout.strip() load = os.getloadavg() process_list = subprocess.run(["ps", "-eo", "args="], text=True, capture_output=True, check=False).stdout.splitlines() busy = [line.strip() for line in process_list if any(token in line for token in ("UnrealBuildTool", "Build.sh", "UnrealEditor-Cmd"))] record = {"uptime": uptime, "load_1_5_15": load, "other_build_or_editor_processes": busy, "checked_utc": utc_now()} if load[0] >= 1.5 or load[1] >= 1.5 or busy: raise RuntimeError(f"machine is not quiet enough for timing: {record}") return record def run_matrix() -> int: builds_path = OUT / "builds.json" if not builds_path.is_file(): raise RuntimeError("missing build manifests; run this script prepare first") builds = json.loads(builds_path.read_text(encoding="utf-8")) fixed_root = Path(builds["fixed"]["plugin_root"]) baseline_root = Path(builds["ca0bf76"]["plugin_root"]) fixed_binary = Path(builds["fixed"]["binary"]) baseline_binary = Path(builds["ca0bf76"]["binary"]) for build in (builds["fixed"], builds["ca0bf76"]): if not build.get("compiled_voxel_world_cpp"): raise RuntimeError("build manifest lacks a full VoxelWorld.cpp compile") sizes = {item["byte_size"] for item in build["dwarf_pre_run_gate"]["objects"].values()} if len(sizes) != 1: raise RuntimeError(f"saved DWARF gate failed: {build['plugin_root']} {sizes}") if sha256_file(fixed_binary) != builds["fixed"]["binary_sha256"]: raise RuntimeError("fixed binary does not match its build manifest") if sha256_file(baseline_binary) != builds["ca0bf76"]["binary_sha256"]: raise RuntimeError("ca0bf76 binary does not match its build manifest") quiet = require_quiet_machine() write_json(OUT / "timing-machine.json", quiet) all_rows: dict[str, list[dict[str, Any]]] = { "aa": load_partial_rows("aa"), "off_vs_ca0bf76": load_partial_rows("off-vs-ca0bf76"), "on_vs_off": load_partial_rows("on-vs-off"), } try: # A/A arms use the exact same staged fixed binary and the same CVar values. aa = all_rows["aa"] for arm in ("aa-a", "aa-b"): if not has_run(aa, arm, "warmup"): run_batch_arm(aa, "aa", arm, fixed_root, fixed_binary, WAIT_OFF, APPLY_OFF_MS, "warmup") for pair in range(1, 11): for condition in ("aa-a", "aa-b"): if not has_run(aa, condition, f"pair-{pair}"): run_batch_arm(aa, "aa", condition, fixed_root, fixed_binary, WAIT_OFF, APPLY_OFF_MS, f"pair-{pair}") aa_pairs = paired_deltas(aa, "aa-a", "aa-b", "frame_resolution_arrival_s") aa_abs_p90_ms = 1000.0 * percentile_nearest_rank( [abs(item["delta_s_second_minus_first"]) for item in aa_pairs], 0.90) aa_summary = {"pairs": aa_pairs, "pair_count": len(aa_pairs), "absolute_delta_p90_ms": aa_abs_p90_ms} write_json(UE_OUT / "aa" / "summary.json", aa_summary) off_vs_base = all_rows["off_vs_ca0bf76"] for condition, root, binary in (("fixed-off", fixed_root, fixed_binary), ("ca0bf76", baseline_root, baseline_binary)): if not has_run(off_vs_base, condition, "warmup"): run_batch_arm(off_vs_base, "off-vs-ca0bf76", condition, root, binary, WAIT_OFF, APPLY_OFF_MS, "warmup") for pair in range(1, 11): control_order = (("fixed-off", "ca0bf76") if pair % 2 else ("ca0bf76", "fixed-off")) for condition in control_order: if has_run(off_vs_base, condition, f"pair-{pair}"): continue root, binary = ((fixed_root, fixed_binary) if condition == "fixed-off" else (baseline_root, baseline_binary)) run_batch_arm(off_vs_base, "off-vs-ca0bf76", condition, root, binary, WAIT_OFF, APPLY_OFF_MS, f"pair-{pair}") control_pairs = paired_deltas(off_vs_base, "ca0bf76", "fixed-off", "frame_resolution_arrival_s") control_slower = sum(item["delta_s_second_minus_first"] > 0 for item in control_pairs) control_summary = {"pairs": control_pairs, "pair_count": len(control_pairs), "fixed_off_slower_pairs": control_slower, "stop_at_eight_slower": control_slower >= 8} write_json(UE_OUT / "off-vs-ca0bf76" / "summary.json", control_summary) if control_slower >= 8: final = {"status": "STOPPED_CONTROL_FAILED", "timing_machine": quiet, "aa": aa_summary, "fixed_off_vs_ca0bf76": control_summary, "fixed_on_vs_off": None, "reason": "fixed-off was slower than true ca0bf76 in at least 8/10 pairs"} write_json(UE_OUT / "decision-summary.json", final) print("[L172Decision] " + json.dumps({key: value for key, value in final.items() if key not in ("aa", "fixed_off_vs_ca0bf76")}, sort_keys=True), flush=True) return 0 on_vs_off = all_rows["on_vs_off"] for condition, wait, budget in (("fixed-on", WAIT_ON, APPLY_ON_MS), ("fixed-off", WAIT_OFF, APPLY_OFF_MS)): if not has_run(on_vs_off, condition, "warmup"): run_batch_arm(on_vs_off, "on-vs-off", condition, fixed_root, fixed_binary, wait, budget, "warmup") settings = {"fixed-on": (WAIT_ON, APPLY_ON_MS), "fixed-off": (WAIT_OFF, APPLY_OFF_MS)} for pair in range(1, 11): on_off_order = (("fixed-on", "fixed-off") if pair % 2 else ("fixed-off", "fixed-on")) for condition in on_off_order: if has_run(on_vs_off, condition, f"pair-{pair}"): continue wait, budget = settings[condition] run_batch_arm(on_vs_off, "on-vs-off", condition, fixed_root, fixed_binary, wait, budget, f"pair-{pair}") promotion_pairs = paired_deltas(on_vs_off, "fixed-off", "fixed-on", "frame_resolution_arrival_s") arrival_deltas_ms = [1000.0 * item["delta_s_second_minus_first"] for item in promotion_pairs] paired_median_delta_ms = statistics.median(arrival_deltas_ms) slower_pairs = sum(value > 0 for value in arrival_deltas_ms) tolerance_ms = max(aa_abs_p90_ms, 17.0) fixed_off_rows = [row for row in on_vs_off if row["condition"] == "fixed-off" and row["run"].startswith("pair-")] fixed_on_rows = [row for row in on_vs_off if row["condition"] == "fixed-on" and row["run"].startswith("pair-")] near_gain = (statistics.median(row["near_readiness_moving_mean_pct"] for row in fixed_on_rows) - statistics.median(row["near_readiness_moving_mean_pct"] for row in fixed_off_rows)) far_change = (statistics.median(row["far_holes_moving_median_pct"] for row in fixed_on_rows) - statistics.median(row["far_holes_moving_median_pct"] for row in fixed_off_rows)) moving_gate_rows = [row for row in on_vs_off if row["run"].startswith("pair-")] gates = { "arrival_median_delta_within_tolerance": paired_median_delta_ms <= tolerance_ms, "arrival_candidate_slower_in_at_most_6_of_10": slower_pairs <= 6, "near_gain_at_least_9_6_pp": near_gain >= 9.6, "far_hole_change_at_most_minus_1_35_pp": far_change <= -1.35, "on_off_moving_runs_p99_at_most_11_ms": all( row["tick_p99_ms"] is not None and row["tick_p99_ms"] <= 11.0 for row in moving_gate_rows), "on_off_moving_runs_worst_at_most_14_ms": all( row["tick_worst_ms"] is not None and row["tick_worst_ms"] <= 14.0 for row in moving_gate_rows), } on_off_summary = { "pairs": promotion_pairs, "paired_median_delta_ms": paired_median_delta_ms, "candidate_slower_pairs": slower_pairs, "arrival_tolerance_ms": tolerance_ms, "aa_absolute_delta_p90_ms": aa_abs_p90_ms, "near_gain_pp": near_gain, "far_hole_change_pp": far_change, "gates": gates, } write_json(UE_OUT / "on-vs-off" / "summary.json", on_off_summary) promote = all(gates.values()) and not control_summary["stop_at_eight_slower"] final = { "status": "PROMOTE" if promote else "KEEP_DEFAULTS_OFF", "timing_machine": quiet, "aa": aa_summary, "fixed_off_vs_ca0bf76": control_summary, "fixed_on_vs_off": on_off_summary, "moving_metrics_medians": { "fixed_off_near_pct": statistics.median( row["near_readiness_moving_mean_pct"] for row in fixed_off_rows), "fixed_on_near_pct": statistics.median( row["near_readiness_moving_mean_pct"] for row in fixed_on_rows), "fixed_off_far_holes_pct": statistics.median( row["far_holes_moving_median_pct"] for row in fixed_off_rows), "fixed_on_far_holes_pct": statistics.median( row["far_holes_moving_median_pct"] for row in fixed_on_rows), "fixed_on_p99_ms_max": max(row["tick_p99_ms"] for row in fixed_on_rows), "fixed_on_worst_tick_ms_max": max(row["tick_worst_ms"] for row in fixed_on_rows), }, "promotion_gates": gates, "all_runs": all_rows, "switch_defaults": { "voxel.StreamingFastTravelWaitForAllCriticalTiles": 0 if promote else 1, "voxel.StreamingFastTravelApplyBudgetMilliseconds": 1.0 if promote else 0.0, "voxel.StreamingIncrementalDesiredSet": 1, "voxel.StreamingFastTravelForwardFirst": 0, "voxel.StreamingRebuildOnCoreRegionCompletion": 1, "voxel.StreamingCoreDesiredCacheByCenter": 0, "voxel.StreamingIncrementalDesiredAuditEvery": 0, "voxel.StreamingIncrementalDesiredMaterializationAuditEvery": 0, "voxel.StreamingIncrementalDesiredAuditInjectMismatch": 0, }, } write_json(UE_OUT / "decision-summary.json", final) print("[L172Decision] " + json.dumps({ "status": final["status"], "aa_abs_p90_ms": aa_abs_p90_ms, "arrival_median_delta_ms": paired_median_delta_ms, "arrival_tolerance_ms": tolerance_ms, "slower_pairs": slower_pairs, "near_gain_pp": near_gain, "far_hole_change_pp": far_change, "gates": gates}, sort_keys=True), flush=True) return 0 finally: restored = restore_main_checkout() write_json(OUT / "restored-main-links.json", { "time_utc": utc_now(), "main_root": str(MAIN_CHECKOUT.resolve()), "links": restored["links"], "intermediates_moved": restored["intermediates_moved"]}) def run_default_smoke() -> int: build = json.loads((OUT / "build-fixed-default-on.json").read_text(encoding="utf-8")) try: row = run_trial("default-smoke", "default-on-smoke", "fixed-on", ROOT, Path(build["binary"]), WAIT_ON, APPLY_ON_MS, include_switch_overrides=False) write_json(UE_OUT / "default-on-smoke" / "summary.json", row) print("[L172DefaultSmoke] " + json.dumps(row, sort_keys=True), flush=True) return 0 finally: restored = restore_main_checkout() write_json(OUT / "restored-main-links.json", { "time_utc": utc_now(), "main_root": str(MAIN_CHECKOUT.resolve()), "links": restored["links"], "intermediates_moved": restored["intermediates_moved"]}) def main() -> int: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("prepare", help="make a fresh ca0bf76 worktree and clean UE builds") subparsers.add_parser("rescore", help="rescore saved L169 and L170 traces by scheduler frame") subparsers.add_parser("matrix", help="run the L172 A/A and counterbalanced 10-pair matrices") subparsers.add_parser("default-smoke", help="run the built default-on binary without CVar overrides") args = parser.parse_args() if args.command == "prepare": return prepare() if args.command == "rescore": rescore_saved_traces() return 0 if args.command == "matrix": return run_matrix() if args.command == "default-smoke": return run_default_smoke() return 2 if __name__ == "__main__": raise SystemExit(main())