#!/usr/bin/env python3 """Interleaved L157 UE movement runs for critical-gate and apply-budget ablations.""" from __future__ import annotations import argparse import json import math import statistics import subprocess import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] HOST = Path("/home/jahni/VFHost") EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd") OUT = ROOT / "Saved/Linux/l157/ue/movement" HOST_SAVED = HOST / "Saved/Linux/l157" HOST_PLUGIN_SAVED = HOST / "Plugins/VoxelForge/Saved/Linux/l157" SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375) MOVE_START_S = 6.0 CONDITIONS = { "assembler_world": {"wait_for_critical": 1, "apply_budget_ms": 0.0}, "early_admission_apply_1ms": {"wait_for_critical": 0, "apply_budget_ms": 1.0}, } ORDER = { 1: ("assembler_world", "early_admission_apply_1ms"), 2: ("early_admission_apply_1ms", "assembler_world"), 3: ("assembler_world", "early_admission_apply_1ms"), 4: ("early_admission_apply_1ms", "assembler_world"), 5: ("assembler_world", "early_admission_apply_1ms"), } def percentile(values: list[float], fraction: float = 0.99) -> float | None: if not values: return None ordered = sorted(values) return ordered[max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1))] def read_events(path: Path) -> list[dict]: rows = [] if path.is_file(): 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 mean(values: list[float]) -> float | None: return statistics.mean(values) if values else None def percent(numerator: int | float, denominator: int | float) -> float | None: return 100.0 * numerator / denominator if denominator > 0 else None def run(condition: str, repetition: int, tile_hash_dump: Path | None = None) -> dict: config = CONDITIONS[condition] run_name = f"fast-r{repetition}" out = OUT / condition host_out = HOST_PLUGIN_SAVED / condition / "output" userdir = HOST_SAVED / "user" / condition / run_name zendata = HOST_SAVED / "zen" / condition / run_name ddc = HOST_SAVED / "DDC" trace = out / f"{run_name}.trace.jsonl" abslog = out / f"{run_name}.log" stdout = out / f"{run_name}.stdout.log" export = host_out / run_name for directory in (out, export, userdir, zendata, ddc): directory.mkdir(parents=True, exist_ok=True) 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.StreamingCoarseFirst=1", "-voxel.StreamingDynamicNearReserve=1", "-voxel.StreamingLegacyPendingResultCap=0", "-voxel.CoreRegionConcurrentAdmission=1", "-voxel.StreamingResultBacklogCap=256", "-voxel.StreamingReadyQueueDepth=8", "-voxel.StreamingBackgroundWorkerCap=1", "-voxel.CoreColumnCacheDirectMapped=0", "-voxel.StreamingFarApplySupplement=1", f"-voxel.StreamingFastTravelWaitForAllCriticalTiles={config['wait_for_critical']}", f"-voxel.StreamingFastTravelApplyBudgetMilliseconds={config['apply_budget_ms']:.1f}", "-voxel.CoreMCGradientMemoization=1", "-voxel.StreamingCollisionLookahead=1", "-voxel.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=3000.0", f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}", "-voxel.TestMoveDurationSeconds=12.0", "-voxel.TestExitSeconds=23.0", "-voxel.TestMoveDirection=1,0,0", "-voxel.StreamingSchedulerPolicy=3", "-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={abslog}", ] if tile_hash_dump is not None: tile_hash_dump.parent.mkdir(parents=True, exist_ok=True) tile_hash_dump.unlink(missing_ok=True) command.append(f"-voxel.TileHashDump={tile_hash_dump.resolve()}") wrapped = ["taskset", "-c", "0-11", "timeout", "1740s", *command] started = time.monotonic() with stdout.open("w", encoding="utf-8") as stream: process = subprocess.run( wrapped, cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740) events = read_events(trace) samples = [event for event in events if event.get("type") == "streaming_sample"] moving = [event for event in samples if float(event.get("speed_cm_s", 0.0)) >= 50.0] stop_event = next((event for event in events if event.get("type") == "streaming_test_move_stopped"), None) backend = next((event for event in events if event.get("type") == "worker_backend_config"), {}) worker_cap = next((event for event in events if event.get("type") == "streaming_worker_cap_readback"), {}) if stop_event is None: raise RuntimeError(f"missing move-stop event: {trace}") stop_s = float(stop_event.get("t_s", 0.0)) move_start_s = min((float(event.get("t_s", 0.0)) for event in moving), default=float("inf")) 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] arrival = [event for event in samples if float(event.get("t_s", 0.0)) < move_start_s and float(event.get("speed_cm_s", 0.0)) < 50.0] if process.returncode or len(moving) < 40 or len(dwell) < 16: raise RuntimeError(f"incomplete {condition} run: {trace}; exit={process.returncode}; " f"move={len(moving)} dwell={len(dwell)}") if (int(backend.get("streaming_fast_travel_wait_for_all_critical_tiles", -1)) != config["wait_for_critical"] or abs(float(backend.get("streaming_fast_travel_apply_budget_ms", -1.0)) - config["apply_budget_ms"]) > 0.01 or int(worker_cap.get("streaming_background_worker_cap", -1)) != 1 or int(backend.get("effective_tile_slots", -1)) != 8 or int(backend.get("streaming_far_apply_supplement", -1)) != 1): raise RuntimeError(f"CVar/worker readback mismatch: {backend}; {worker_cap}") moving_metrics = [] for event in moving: desired = event.get("desired_by_lod", [0] * 9) resident = event.get("resident_by_lod", [0] * 9) near_desired = sum(desired[:3]) cone_desired = int(event.get("near_forward_cone_desired", 0)) weighted_desired = float(event.get("near_forward_weighted_desired", 0.0)) moving_metrics.append({ "near_ready_pct": percent(sum(resident[:3]), near_desired) or 0.0, "lod0_ready_pct": percent(resident[0], desired[0]) or 0.0, "lod1_ready_pct": percent(resident[1], desired[1]) or 0.0, "lod2_ready_pct": percent(resident[2], desired[2]) or 0.0, "far_hole_pct": percent( sum(desired[3:]) - sum(event.get("covered_by_lod", resident)[3:]), sum(desired[3:])) or 0.0, "ahead_share_pct": percent(int(event.get("near_ahead_desired", 0)), near_desired), "behind_share_pct": percent(int(event.get("near_behind_desired", 0)), near_desired), "side_share_pct": 100.0 - ( percent(int(event.get("near_ahead_desired", 0)), near_desired) or 0.0 ) - ( percent(int(event.get("near_behind_desired", 0)), near_desired) or 0.0 ), "ahead_ready_pct": percent( int(event.get("near_ahead_resident", 0)), int(event.get("near_ahead_desired", 0))), "behind_ready_pct": percent( int(event.get("near_behind_resident", 0)), int(event.get("near_behind_desired", 0))), "forward_cone_ready_pct": percent( int(event.get("near_forward_cone_resident", 0)), cone_desired), "forward_cone_share_pct": percent(cone_desired, near_desired), "forward_weighted_ready_pct": percent( float(event.get("near_forward_weighted_resident", 0.0)), weighted_desired), }) arrival_near_settled = next((event for event in arrival if sum(event.get("desired_by_lod", [0] * 9)[:3]) > 0 and sum(event.get("settled_by_lod", [0] * 9)[:3]) >= sum(event.get("desired_by_lod", [0] * 9)[:3])), None) dwell_complete = next((event for event in dwell if sum(event.get("resident_by_lod", [0] * 9)[:3]) >= sum(event.get("desired_by_lod", [0] * 9)[:3]) and sum(event.get("desired_by_lod", [0] * 9)[3:]) <= sum(event.get("covered_by_lod", [0] * 9)[3:])), None) work = [event for event in events if event.get("type") == "game_thread_work_frame" and move_start_s <= float(event.get("t_s", 0.0)) < stop_s] applies = {event.get("frame"): event for event in events if event.get("type") == "apply_frame" and move_start_s <= float(event.get("t_s", 0.0)) < stop_s} scheduler = {event.get("frame"): event for event in events if event.get("type") == "scheduler_admission" and move_start_s <= float(event.get("t_s", 0.0)) < stop_s} frames = [] for event in work: apply = applies.get(event.get("frame"), {}) admit = scheduler.get(event.get("frame"), {}) frames.append({ "frame": event.get("frame"), "tick_ms": float(event.get("tick_ms", 0.0)), "update_chunks_ms": float(event.get("update_chunks_ms", 0.0)), "pending_drain_ms": float(event.get("pending_drain_ms", 0.0)), "apply_ms": float(apply.get("apply_ms", 0.0)), "apply_budget_ms": float(apply.get("apply_budget_ms", 0.0)), "meshes_applied": int(apply.get("meshes_applied", 0)), "time_budget_hit": int(apply.get("time_limit_reached", 0)), "desired_rebuild_ms": float(admit.get("desired_rebuild_ms", 0.0)), "submit_tiles_ms": float(admit.get("submit_tiles_ms", 0.0)), "level_admit_ms": float(admit.get("level_admit_ms", 0.0)), }) p99_field = lambda key: percentile([float(row[key]) for row in frames]) max_field = lambda key: max((float(row[key]) for row in frames), default=None) dwell_complete_s = (float(dwell_complete.get("t_s", 0.0)) - stop_s if dwell_complete else None) row = { "condition": condition, "repetition": repetition, "speed_mps": 30.0, "wait_for_all_critical_tiles": config["wait_for_critical"], "fast_travel_apply_budget_ms": config["apply_budget_ms"], "effective_tile_slots": backend.get("effective_tile_slots"), "exit_code": process.returncode, "wall_s": round(time.monotonic() - started, 3), "moving_samples": len(moving), "dwell_samples": len(dwell), "arrival_samples": len(arrival), "near_ready_moving_mean_pct": mean([m["near_ready_pct"] for m in moving_metrics]), "lod0_ready_moving_mean_pct": mean([m["lod0_ready_pct"] for m in moving_metrics]), "lod1_ready_moving_mean_pct": mean([m["lod1_ready_pct"] for m in moving_metrics]), "lod2_ready_moving_mean_pct": mean([m["lod2_ready_pct"] for m in moving_metrics]), "far_holes_moving_mean_pct": mean([m["far_hole_pct"] for m in moving_metrics]), "near_ahead_desired_share_pct": mean([m["ahead_share_pct"] for m in moving_metrics if m["ahead_share_pct"] is not None]), "near_behind_desired_share_pct": mean([m["behind_share_pct"] for m in moving_metrics if m["behind_share_pct"] is not None]), "near_side_desired_share_pct": mean([m["side_share_pct"] for m in moving_metrics]), "near_ahead_ready_pct": mean([m["ahead_ready_pct"] for m in moving_metrics if m["ahead_ready_pct"] is not None]), "near_behind_ready_pct": mean([m["behind_ready_pct"] for m in moving_metrics if m["behind_ready_pct"] is not None]), "near_forward_cone_half_angle_deg": 55, "near_forward_cone_desired_share_pct": mean([m["forward_cone_share_pct"] for m in moving_metrics if m["forward_cone_share_pct"] is not None]), "near_forward_cone_ready_pct": mean([m["forward_cone_ready_pct"] for m in moving_metrics if m["forward_cone_ready_pct"] is not None]), "near_forward_cosine_weighted_ready_pct": mean( [m["forward_weighted_ready_pct"] for m in moving_metrics if m["forward_weighted_ready_pct"] is not None]), "arrival_lod02_settled_t_s": ( float(arrival_near_settled.get("t_s", 0.0)) if arrival_near_settled else None), "near_ready_dwell_end_pct": (100.0 * sum(dwell[-1].get("resident_by_lod", [0] * 9)[:3]) / max(1, sum(dwell[-1].get("desired_by_lod", [0] * 9)[:3]))), "far_holes_dwell_end_pct": (100.0 * ( sum(dwell[-1].get("desired_by_lod", [0] * 9)[3:]) - sum(dwell[-1].get("covered_by_lod", [0] * 9)[3:])) / max(1, sum(dwell[-1].get("desired_by_lod", [0] * 9)[3:]))), "dwell_complete_after_stop_s": dwell_complete_s, "game_thread_tick_p99_ms": p99_field("tick_ms"), "game_thread_tick_max_ms": max_field("tick_ms"), "stream_update_p99_ms": p99_field("update_chunks_ms"), "pending_drain_p99_ms": p99_field("pending_drain_ms"), "apply_p99_ms": p99_field("apply_ms"), "apply_max_ms": max_field("apply_ms"), "apply_budget_min_ms": min((f["apply_budget_ms"] for f in frames), default=None), "apply_budget_max_ms": max((f["apply_budget_ms"] for f in frames), default=None), "apply_time_budget_hit_frames": sum(f["time_budget_hit"] for f in frames), "apply_frames": sum(f["meshes_applied"] > 0 for f in frames), "apply_meshes_total": sum(f["meshes_applied"] for f in frames), "desired_rebuild_p99_ms": p99_field("desired_rebuild_ms"), "submit_tiles_p99_ms": p99_field("submit_tiles_ms"), "level_admit_p99_ms": p99_field("level_admit_ms"), "top_tick_frames": sorted(frames, key=lambda f: f["tick_ms"], reverse=True)[:5], "trace": str(trace.relative_to(ROOT)), "log": str(abslog.relative_to(ROOT)), } print("[L157Movement] " + json.dumps(row, sort_keys=True), flush=True) return row def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--smoke", action="store_true", help="run one candidate trace to validate CVar readback and event fields") parser.add_argument("--geometry-hash-check", action="store_true", help="compare UE per-tile geometry hashes for baseline and early admission") args = parser.parse_args() missing = [path for path in (EDITOR, HOST / "VFHost.uproject") if not path.exists()] if missing: raise SystemExit(f"missing UE runner input: {missing}") OUT.mkdir(parents=True, exist_ok=True) if args.geometry_hash_check: hash_dir = OUT / "geometry-hash" baseline_path = hash_dir / "assembler-world.json" candidate_path = hash_dir / "early-admission-apply-1ms.json" run("assembler_world", 99, baseline_path) run("early_admission_apply_1ms", 99, candidate_path) baseline_data = json.loads(baseline_path.read_text(encoding="utf-8")) candidate_data = json.loads(candidate_path.read_text(encoding="utf-8")) key_fields = ("level", "tile_x", "tile_y", "tile_z") def indexed(data: dict) -> dict[tuple[int, ...], tuple]: result = {} for record in data.get("records", []): key = tuple(int(record[field]) for field in key_fields) result[key] = (str(record.get("geometry_hash", "")), int(record.get("triangle_count", -1)), bool(record.get("empty", False))) return result baseline_hashes = indexed(baseline_data) candidate_hashes = indexed(candidate_data) common = baseline_hashes.keys() & candidate_hashes.keys() mismatches = [key for key in common if baseline_hashes[key] != candidate_hashes[key]] result = { "baseline_records": len(baseline_hashes), "candidate_records": len(candidate_hashes), "common_records": len(common), "hash_mismatches": len(mismatches), "baseline_only_records": len(baseline_hashes.keys() - candidate_hashes.keys()), "candidate_only_records": len(candidate_hashes.keys() - baseline_hashes.keys()), "baseline_dump": str(baseline_path.relative_to(ROOT)), "candidate_dump": str(candidate_path.relative_to(ROOT)), } (hash_dir / "comparison.json").write_text( json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") print("[L157GeometryHash] " + json.dumps(result, sort_keys=True), flush=True) return 0 if len(common) >= 100 and not mismatches else 1 if args.smoke: run("early_admission_apply_1ms", 0) return 0 rows = [] partial = OUT / "runs-partial.json" for repetition in range(1, 6): for condition in ORDER[repetition]: row = run(condition, repetition) rows.append(row) partial.write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n", encoding="utf-8") metrics = ( "near_ready_moving_mean_pct", "far_holes_moving_mean_pct", "game_thread_tick_p99_ms", "game_thread_tick_max_ms", "arrival_lod02_settled_t_s", "dwell_complete_after_stop_s", "near_ahead_desired_share_pct", "near_behind_desired_share_pct", "near_side_desired_share_pct", "near_forward_cone_desired_share_pct", "near_ahead_ready_pct", "near_behind_ready_pct", "near_forward_cone_ready_pct", "near_forward_cosine_weighted_ready_pct", "apply_p99_ms", "desired_rebuild_p99_ms", "level_admit_p99_ms", ) medians = { condition: {metric: statistics.median( row[metric] for row in rows if row["condition"] == condition) for metric in metrics} for condition in CONDITIONS } pairs = [] for repetition in range(1, 6): by_condition = {row["condition"]: row for row in rows if row["repetition"] == repetition} for candidate in ("early_admission_apply_1ms",): pairs.append({ "repetition": repetition, "candidate": candidate, **{f"{metric}_delta": by_condition[candidate][metric] - by_condition["assembler_world"][metric] for metric in metrics if by_condition[candidate][metric] is not None and by_condition["assembler_world"][metric] is not None}, }) summary = { "source_commit": subprocess.run( ["git", "-c", f"safe.directory={ROOT}", "rev-parse", "HEAD"], cwd=ROOT, check=True, capture_output=True, text=True).stdout.strip(), "protocol": "Five interleaved 30 m/s UE runs per condition; 12 s move plus 5 s dwell; 8 effective tile slots.", "conditions": CONDITIONS, "movement_medians": medians, "paired_deltas_vs_assembler_world": pairs, "runs": rows, } (OUT / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8") print("[L157Summary] " + json.dumps({"movement_medians": medians, "runs": len(rows)}, sort_keys=True), flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())