streaming: latch fast-travel early admission until the arrival set settles (L170, still opt-in)

Switches verified inactive before first settlement; moving gains hold (near 43.7 -> 55.5%, far holes 1.8 -> 0%, p99
7.21 -> 6.48 ms, worst 12.32 -> 12.27 ms). Arrival median +1.28% (4/7 pairs slower) though the switches did no work
then; defaults remain off (WaitForAllCriticalTiles=1, ApplyBudget=0). Source/: VoxelWorld.cpp, VoxelWorld.h.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1

Standalone/Research/WorldAssemblerM3/Linux170/REPORT.md

@@ -0,0 +1,55 @@
+# L170 — Gate early admission on settled arrival
+
+**Result:** The initial-arrival latch works: fast-travel admission and its 1 ms apply budget stayed inactive until the LOD0–2 desired set and required collision set settled, then became active during 30 m/s movement. Promotion still fails the arrival gate, so the source defaults remain off. The candidate was 1.282% slower on the per-condition arrival medians and was slower in four of seven pairs.
+
+## First-arrival census
+
+The UE matrix used seven interleaved W8 pairs against `assembler/world` `ca0bf76`, 30 m/s, 12 seconds of movement plus 5 seconds of dwell, and one excluded warm-up per binary. The candidate binary was compiled with the two early-admission defaults on; there were no CVar command-line overrides. Its raw first-settled LOD0–2 median was 1.691450 s versus 1.670039 s for baseline (+21.411 ms, +1.282%). The median paired delta was +12.176 ms; the candidate was not slower in only three pairs. Times from the first streaming sample to settlement had medians 1.2965 s and 1.3132 s respectively.
+
+I counted startup-window trace timers from each run's first streaming sample through its first settled LOD0–2 sample. Candidate fast-travel mode was active in zero pre-settlement scheduler samples, and its apply budget remained at the normal 2 ms. During that same window:
+
+- Level candidate scanning summed to 2.896 ms baseline and 3.758 ms candidate (paired median delta +0.895 ms). Partial level sorting fell from 2.195 ms to 0.727 ms (−1.452 ms paired median).
+- Desired rebuild totals were effectively unchanged: 3.046 ms versus 3.047 ms per window (paired median delta −0.035 ms). The candidate spent 0.149 ms in its transition-index phase; the baseline has no matching phase.
+- Result-drain time was 127.986 ms versus 131.521 ms (paired median delta +5.001 ms). Apply time was 121.509 ms versus 125.095 ms (+5.416 ms paired median), while the candidate had applied 10 fewer meshes by settlement at the paired median.
+
+These timers locate the largest measured difference in result draining and apply work, but they do not isolate why that work costs more; some phase timers may overlap and should not be added. The early-admission switches themselves did no work before settlement, so the census does not attribute the remaining arrival regression to those switches.
+
+## Movement results
+
+Medians over the seven 30 m/s moving windows:
+
+| Metric | `ca0bf76` | Candidate | Change |
+|---|---:|---:|---:|
+| Near LOD0–2 readiness | 43.65% | 55.46% | +11.82 pp |
+| Forward-cone readiness | 15.79% | 18.69% | +2.90 pp |
+| Far holes | 1.80% | 0.00% | −1.80 pp |
+| Tick p99 | 7.214 ms | 6.476 ms | lower |
+| Worst tick | 12.315 ms | 12.272 ms | lower |
+
+All runs met p99 ≤ 11 ms and worst tick ≤ 14 ms; the largest observed values across both conditions were 7.281 ms and 12.881 ms. Candidate fast-travel mode and the 1 ms budget were active in every moving run. The movement gains met the L169-relative gates; only the two arrival gates failed.
+
+## Defaults and checks
+
+| CVar | Final source default | Candidate test default |
+|---|---:|---:|
+| `voxel.StreamingFastTravelWaitForAllCriticalTiles` | 1 | 0 |
+| `voxel.StreamingFastTravelApplyBudgetMilliseconds` | 0 ms | 1 ms |
+| `voxel.StreamingIncrementalDesiredSet` | 1 | 1 |
+| `voxel.StreamingFastTravelForwardFirst` | 0 | 0 |
+| `voxel.StreamingRebuildOnCoreRegionCompletion` | 1 | 1 |
+| `voxel.StreamingCoreDesiredCacheByCenter` | 0 | 0 |
+| `voxel.StreamingIncrementalDesiredAuditEvery` | 0 | 0 |
+| `voxel.StreamingIncrementalDesiredMaterializationAuditEvery` | 0 | 0 |
+| `voxel.StreamingIncrementalDesiredAuditInjectMismatch` | 0 | 0 |
+
+- VFHost UE Linux Development builds passed for the candidate-on measurement binary and final defaults-off source (`-NoUBA -MaxParallelActions=4`); three existing deprecated RealtimeMesh API warnings remain.
+- `Standalone/Tools/verify/verify.py fast` passed on the final source in 280.7 s: 268 passes, zero failures, five non-gating hash-baseline diagnostics. Worker-count determinism checks passed.
+- The L170 movement runner completed all seven pairs. Its exit was nonzero only because the two arrival promotion gates failed. Runner AST parsing and `git diff --check` passed.
+- UE W12 movement timing was not run; the VFHost procedure reported eight effective tile slots. Fast-tier W1/W8/W12 determinism checks ran.
+
+No geometry or evaluator code changed. Remaining risk is the unexplained arrival regression in result-drain/apply work; promotion should wait for a measurement that isolates that cost.
+
+Changed Unreal `Source/` files:
+
+- `Source/VoxelForge/Private/VoxelWorld.cpp`
+- `Source/VoxelForge/Public/VoxelWorld.h`

Standalone/Research/WorldAssemblerM3/Linux170/RunL170Movement.py

@@ -0,0 +1,490 @@
+#!/usr/bin/env python3
+"""L170 compiled-default UE movement comparison and startup-gate trace."""
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import shutil
+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")
+HOST_PLUGIN = HOST / "Binaries/Linux/libUnrealEditor-VoxelForge.so"
+OUT = ROOT / "Saved/Linux/l170/ue/movement"
+BIN_DIR = ROOT / "Saved/Linux/l170/bin"
+SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375)
+MOVE_START_S = 6.0
+MOVE_DURATION_S = 12.0
+
+BASELINE_BINARY = ROOT / "Saved/Linux/l169/bin/ca0bf76/libUnrealEditor-VoxelForge.so"
+CANDIDATE_BINARY = BIN_DIR / "candidate-default-on/libUnrealEditor-VoxelForge.so"
+DIAGNOSTIC_BINARY = CANDIDATE_BINARY
+
+EXPECTED = {
+    "baseline": {"wait": 1, "apply_budget_ms": 0.0, "incremental": 1},
+    "candidate": {"wait": 0, "apply_budget_ms": 1.0, "incremental": 1},
+}
+
+ORDER = {
+    1: ("baseline", "candidate"),
+    2: ("candidate", "baseline"),
+    3: ("baseline", "candidate"),
+    4: ("candidate", "baseline"),
+    5: ("baseline", "candidate"),
+    6: ("candidate", "baseline"),
+    7: ("baseline", "candidate"),
+}
+
+
+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 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 summarize(condition: str, run_name: str, trace: Path, log: Path, exit_code: int,
+              wall_s: float, audit_or_diag: bool = False) -> dict:
+    expected = EXPECTED[condition]
+    events = read_events(trace)
+    backend = next((e for e in events if e.get("type") == "worker_backend_config"), {})
+    scheduler_all = [e for e in events if e.get("type") == "scheduler_admission"]
+    if not scheduler_all:
+        raise RuntimeError(f"missing scheduler readback: {trace}")
+    if any(int(e.get("incremental_desired_set", -1)) != expected["incremental"]
+           for e in scheduler_all):
+        raise RuntimeError(f"incremental desired-set readback mismatch: {trace}")
+    if any(int(e.get("forward_first", -1)) != 0
+           or int(e.get("rebuild_on_core_region_completion", -1)) != 1
+           or int(e.get("core_desired_cache_by_center", -1)) != 0
+           or int(e.get("incremental_desired_audit_every", -1)) != 0
+           or int(e.get("incremental_desired_materialization_audit_every", -1)) != 0
+           or int(e.get("incremental_desired_audit_inject_mismatch", -1)) != 0
+           for e in scheduler_all):
+        raise RuntimeError(f"non-default L160/audit CVar readback: {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["apply_budget_ms"]) > 0.01
+            or int(backend.get("effective_tile_slots", -1)) != 8):
+        raise RuntimeError(f"early-admission/worker readback mismatch: {trace}; {backend}")
+    if condition == "candidate" and any(
+            int(e.get("wait_for_all_critical_tiles", -1)) != expected["wait"]
+            or "initial_arrival_set_settled" not in e
+            or "fast_travel_mode" not in e
+            for e in scheduler_all):
+        raise RuntimeError(f"missing L170 arrival-gate readback: {trace}")
+
+    move_stop = next((e for e in events if e.get("type") == "streaming_test_move_stopped"), None)
+    if move_stop is None:
+        raise RuntimeError(f"missing move-stop event: {trace}")
+    stop_s = float(move_stop.get("t_s", 0.0))
+    samples = [e for e in events if e.get("type") == "streaming_sample"]
+    moving = [e for e in samples if MOVE_START_S <= float(e.get("t_s", 0.0)) < stop_s
+              and float(e.get("speed_cm_s", 0.0)) >= 50.0]
+    dwell = [e for e in samples if float(e.get("t_s", 0.0)) >= stop_s
+             and float(e.get("speed_cm_s", 0.0)) < 50.0]
+    if exit_code != 0 or len(moving) < 40 or len(dwell) < 16:
+        raise RuntimeError(f"incomplete {run_name}: exit={exit_code}, move={len(moving)}, "
+                           f"dwell={len(dwell)}")
+    move_start = min(float(e.get("t_s", 0.0)) for e in moving)
+
+    work = [e for e in events if e.get("type") == "game_thread_work_frame"
+            and move_start <= float(e.get("t_s", 0.0)) < stop_s]
+    scheduler_moving = [e for e in scheduler_all
+                        if move_start <= float(e.get("t_s", 0.0)) < stop_s]
+    rebuild_events = [e for e in events if e.get("type") == "desired_rebuild_breakdown"
+                      and move_start <= float(e.get("t_s", 0.0)) < stop_s]
+    applies = {int(e.get("frame", -1)): e for e in events
+               if e.get("type") == "apply_frame"
+               and move_start <= float(e.get("t_s", 0.0)) < stop_s}
+    scheduler = {int(e.get("frame", -1)): e for e in scheduler_all
+                 if move_start <= float(e.get("t_s", 0.0)) < stop_s}
+    desired = {int(e.get("frame", -1)): e for e in events
+               if e.get("type") == "desired_rebuild_breakdown"
+               and move_start <= float(e.get("t_s", 0.0)) < stop_s}
+    moving_tiles = [e for e in events if e.get("type") == "tile"
+                    and move_start <= float(e.get("t_s", 0.0)) < stop_s]
+    apply_deferrals = [e for e in events if e.get("type") == "tile_apply_deferred"
+                       and move_start <= float(e.get("t_s", 0.0)) < stop_s]
+
+    frames = []
+    for event in work:
+        frame = int(event.get("frame", -1))
+        apply = applies.get(frame, {})
+        admit = scheduler.get(frame, {})
+        rebuild = desired.get(frame, {})
+        frames.append({
+            "frame": 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)),
+            "desired_rebuild_event_ms": float(rebuild.get("total_ms", 0.0)),
+            "overlay_materialize_ms": float(rebuild.get("overlay_materialize_ms", 0.0)),
+            "core_prepare_ms": float(rebuild.get("core_prepare_ms", 0.0)),
+            "core_delta_ms": float(rebuild.get("core_delta_ms", 0.0)),
+            "priority_sort_ms": float(rebuild.get("priority_sort_ms", 0.0)),
+            "critical_order_ms": float(rebuild.get("critical_order_ms", 0.0)),
+            "collision_stamp_ms": float(rebuild.get("collision_stamp_ms", 0.0)),
+            "transition_index_ms": float(rebuild.get("transition_index_ms", 0.0)),
+            "retained_region_ms": float(rebuild.get("retained_region_ms", 0.0)),
+            "region_reconcile_ms": float(rebuild.get("region_reconcile_ms", 0.0)),
+            "level_sort_ms": float(admit.get("level_sort_ms", 0.0)),
+            "level_candidate_scan_ms": float(admit.get("level_candidate_scan_ms", 0.0)),
+            "submit_tiles_ms": float(admit.get("submit_tiles_ms", 0.0)),
+        })
+
+    near = []
+    cone = []
+    far_holes = []
+    for event in moving:
+        desired_by_lod = event.get("desired_by_lod", [0] * 9)
+        resident_by_lod = event.get("resident_by_lod", [0] * 9)
+        covered_by_lod = event.get("covered_by_lod", resident_by_lod)
+        near.append(percent(sum(resident_by_lod[:3]), sum(desired_by_lod[:3])) or 0.0)
+        cone.append(percent(int(event.get("near_forward_cone_resident", 0)),
+                            int(event.get("near_forward_cone_desired", 0))) or 0.0)
+        far_holes.append(percent(sum(desired_by_lod[3:]) - sum(covered_by_lod[3:]),
+                                 sum(desired_by_lod[3:])) or 0.0)
+    settled = next((e for e in samples if float(e.get("t_s", 0.0)) < move_start
+                    and float(e.get("speed_cm_s", 0.0)) < 50.0
+                    and sum(e.get("settled_by_lod", [0] * 9)[:3])
+                        >= sum(e.get("desired_by_lod", [0] * 9)[:3]) > 0), None)
+    if settled is None:
+        raise RuntimeError(f"missing initial LOD0-2 settled sample: {trace}")
+    arrival_settle_s = float(settled["t_s"])
+    arrival_scheduler = [e for e in scheduler_all
+                         if float(e.get("t_s", 0.0)) <= arrival_settle_s]
+    apply_all = [e for e in events if e.get("type") == "apply_frame"]
+    if condition == "candidate":
+        if any(int(e.get("fast_travel_mode", 0)) != 0 for e in arrival_scheduler):
+            raise RuntimeError(f"fast-travel admission activated before initial arrival settled: {trace}")
+        if any(float(e.get("apply_budget_ms", 0.0)) < 1.99
+               and float(e.get("t_s", 0.0)) <= arrival_settle_s for e in apply_all):
+            raise RuntimeError(f"fast-travel apply cap activated before initial arrival settled: {trace}")
+        if not any(int(e.get("initial_arrival_set_settled", 0)) == 1
+                   and int(e.get("fast_travel_mode", 0)) == 1
+                   for e in scheduler_moving):
+            raise RuntimeError(f"fast-travel mode never activated after initial arrival settled: {trace}")
+        if not any(float(e.get("apply_budget_ms", 0.0)) <= 1.001
+                   for e in apply_all
+                   if move_start <= float(e.get("t_s", 0.0)) < stop_s):
+            raise RuntimeError(f"compiled 1 ms apply cap never activated while moving: {trace}")
+    p99 = percentile([r["tick_ms"] for r in frames])
+    worst = max(frames, key=lambda r: r["tick_ms"], default={})
+    row = {
+        "condition": condition,
+        "run": run_name,
+        "diagnostic_override": audit_or_diag,
+        "source_default_incremental": expected["incremental"],
+        "early_admission_wait_for_critical": expected["wait"],
+        "early_admission_apply_budget_ms": expected["apply_budget_ms"],
+        "speed_mps": 30,
+        "effective_tile_slots": backend.get("effective_tile_slots"),
+        "moving_samples": len(moving),
+        "dwell_samples": len(dwell),
+        "tick_p99_ms": p99,
+        "tick_worst_ms": worst.get("tick_ms"),
+        "worst_frame": worst,
+        "near_readiness_moving_mean_pct": statistics.mean(near) if near else None,
+        "forward_cone_readiness_moving_mean_pct": statistics.mean(cone) if cone else None,
+        "far_holes_moving_median_pct": statistics.median(far_holes) if far_holes else None,
+        "arrival_lod02_settled_s": float(settled.get("t_s", 0.0)) if settled else None,
+        "arrival_fast_travel_modes_before_settle": sum(
+            int(e.get("fast_travel_mode", 0)) != 0 for e in arrival_scheduler),
+        "arrival_gate_was_settled_before_move": any(
+            int(e.get("initial_arrival_set_settled", 0)) != 0
+            for e in scheduler_all if float(e.get("t_s", 0.0)) < move_start),
+        "moving_fast_travel_mode_frames": sum(
+            int(e.get("fast_travel_mode", 0)) != 0 for e in scheduler_moving),
+        "moving_one_ms_apply_budget_frames": sum(
+            float(e.get("apply_budget_ms", 0.0)) <= 1.001
+            for e in apply_all if move_start <= float(e.get("t_s", 0.0)) < stop_s),
+        "desired_rebuild_count": len(rebuild_events),
+        "incremental_materialization_updates": sum(
+            int(e.get("materialization_update", 0)) != 0 for e in events
+            if e.get("type") == "desired_rebuild_breakdown"
+            and move_start <= float(e.get("t_s", 0.0)) < stop_s),
+        "core_incremental_deltas": sum(
+            int(e.get("incremental_delta", 0)) != 0 for e in events
+            if e.get("type") == "core_desired_cache"
+            and move_start <= float(e.get("t_s", 0.0)) < stop_s),
+        "level_sort_samples_over_1ms": sum(
+            float(e.get("level_sort_ms", 0.0)) > 1.0 for e in scheduler_moving),
+        "level_sort_max_ms": max((float(e.get("level_sort_ms", 0.0))
+                                   for e in scheduler_moving), default=0.0),
+        "desired_rebuild_events_over_5ms": sum(
+            float(e.get("total_ms", 0.0)) > 5.0 for e in rebuild_events),
+        "desired_rebuild_max_ms": max(
+            (float(e.get("total_ms", 0.0)) for e in rebuild_events), default=0.0),
+        "overlay_materialize_events_over_2ms": sum(
+            float(e.get("overlay_materialize_ms", 0.0)) > 2.0 for e in rebuild_events),
+        "overlay_materialize_max_ms": max(
+            (float(e.get("overlay_materialize_ms", 0.0)) for e in rebuild_events), default=0.0),
+        "desired_subphase_max_ms": {
+            key: max((float(e.get(key, 0.0)) for e in rebuild_events), default=0.0)
+            for key in ("core_prepare_ms", "overlay_materialize_ms", "priority_sort_ms",
+                        "critical_order_ms", "collision_stamp_ms", "transition_index_ms",
+                        "retained_region_ms", "region_reconcile_ms")},
+        "apply_budget_hit_frames": sum(f["time_budget_hit"] for f in frames),
+        "apply_frame_count": sum(f["meshes_applied"] > 0 for f in frames),
+        "apply_ms_max": max((f["apply_ms"] for f in frames), default=0.0),
+        "tiles_over_1ms_apply": sum(float(e.get("apply_s", 0.0)) > 0.001
+                                    for e in moving_tiles),
+        "apply_preflight_deferred_count": len(apply_deferrals),
+        "apply_preflight_deferred_max_estimate_ms": max(
+            (float(e.get("estimated_apply_ms", 0.0)) for e in apply_deferrals), default=0.0),
+        "apply_preflight_deferred_max_triangles": max(
+            (int(e.get("triangles", 0)) for e in apply_deferrals), default=0),
+        "largest_tile_applies": sorted(({
+            "apply_ms": float(e.get("apply_s", 0.0)) * 1000.0,
+            "triangles": int(e.get("triangles", 0)),
+            "vertices": int(e.get("vertices", 0)),
+            "level": int(e.get("level", -1)),
+        } for e in moving_tiles),
+            key=lambda e: e["apply_ms"], reverse=True)[:8],
+        "top_ticks": sorted(frames, key=lambda r: r["tick_ms"], reverse=True)[:8],
+        "trace": str(trace.relative_to(ROOT)),
+        "log": str(log.relative_to(ROOT)),
+        "wall_s": round(wall_s, 3),
+    }
+    return row
+
+
+def run(condition: str, run_name: str, diagnostic_override: bool = False,
+        tile_hash_dump: Path | None = None) -> dict:
+    if condition == "baseline":
+        binary = BASELINE_BINARY
+    elif condition == "candidate":
+        binary = CANDIDATE_BINARY
+    else:
+        raise ValueError(condition)
+    if not binary.is_file():
+        raise FileNotFoundError(binary)
+    HOST_PLUGIN.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(binary, HOST_PLUGIN)
+    out = OUT / condition
+    export = HOST / "Plugins/VoxelForge/Saved/Linux/l170" / condition / run_name
+    userdir = HOST / "Saved/Linux/l170/user" / condition / run_name
+    zendata = HOST / "Saved/Linux/l170/zen" / condition / run_name
+    ddc = HOST / "Saved/Linux/l170/DDC"
+    trace = out / f"{run_name}.trace.jsonl"
+    log = out / f"{run_name}.log"
+    stdout = out / f"{run_name}.stdout.log"
+    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.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 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()}")
+    started = time.monotonic()
+    with stdout.open("w", encoding="utf-8") as stream:
+        process = subprocess.run(
+            ["taskset", "-c", "0-11", "timeout", "1740s", *command],
+            cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740)
+    return summarize(condition, run_name, trace, log, process.returncode,
+                     time.monotonic() - started, diagnostic_override)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    modes = parser.add_mutually_exclusive_group(required=True)
+    modes.add_argument("--diagnostic", action="store_true",
+                       help="one candidate trace with compiled defaults and no command-line switch overrides")
+    modes.add_argument("--matrix", action="store_true",
+                       help="warm both binaries, then run seven interleaved pairs with compiled defaults and no switch overrides")
+    modes.add_argument("--geometry-hash-check", action="store_true",
+                       help="compare shared tile hashes from the compiled-default binaries")
+    args = parser.parse_args()
+    missing = [p for p in (EDITOR, HOST / "VFHost.uproject", ROOT / "VoxelForge.uplugin")
+               if not p.exists()]
+    if missing:
+        raise SystemExit(f"missing UE runner input: {missing}")
+    OUT.mkdir(parents=True, exist_ok=True)
+
+    if args.diagnostic:
+        warmup = run("candidate", "diagnostic-warmup", diagnostic_override=True)
+        row = run("candidate", "l170-default-diagnostic", diagnostic_override=True)
+        row["warmup"] = warmup
+        path = OUT / "diagnostic.json"
+        path.write_text(json.dumps(row, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        print("[L169Diagnostic] " + json.dumps(row, sort_keys=True), flush=True)
+        return 0
+
+    if args.geometry_hash_check:
+        hash_dir = OUT / "geometry-hash"
+        baseline_dump = hash_dir / "baseline.json"
+        candidate_dump = hash_dir / "candidate.json"
+        run("baseline", "geometry-hash", tile_hash_dump=baseline_dump)
+        run("candidate", "geometry-hash", tile_hash_dump=candidate_dump)
+        baseline_data = json.loads(baseline_dump.read_text(encoding="utf-8"))
+        candidate_data = json.loads(candidate_dump.read_text(encoding="utf-8"))
+        key_fields = ("level", "tile_x", "tile_y", "tile_z")
+
+        def indexed(data: dict) -> dict[tuple[int, ...], tuple]:
+            return {tuple(int(record[field]) for field in key_fields): (
+                        str(record.get("geometry_hash", "")),
+                        int(record.get("triangle_count", -1)),
+                        bool(record.get("empty", False)))
+                    for record in data.get("records", [])}
+
+        base = indexed(baseline_data)
+        candidate = indexed(candidate_data)
+        common = base.keys() & candidate.keys()
+        mismatches = [key for key in common if base[key] != candidate[key]]
+        result = {"baseline_records": len(base), "candidate_records": len(candidate),
+                  "common_records": len(common), "hash_mismatches": len(mismatches),
+                  "baseline_only_records": len(base.keys() - candidate.keys()),
+                  "candidate_only_records": len(candidate.keys() - base.keys()),
+                  "baseline_dump": str(baseline_dump.relative_to(ROOT)),
+                  "candidate_dump": str(candidate_dump.relative_to(ROOT))}
+        (hash_dir / "comparison.json").write_text(
+            json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        print("[L169GeometryHash] " + json.dumps(result, sort_keys=True), flush=True)
+        return 0 if len(common) >= 100 and not mismatches else 1
+
+    warmups = [run("baseline", "warmup"), run("candidate", "warmup")]
+    rows = []
+    partial = OUT / "runs-partial.json"
+    for pair, sequence in ORDER.items():
+        for condition in sequence:
+            rows.append(run(condition, f"pair-{pair}"))
+            partial.write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n",
+                               encoding="utf-8")
+    metrics = ("tick_p99_ms", "tick_worst_ms", "near_readiness_moving_mean_pct",
+               "forward_cone_readiness_moving_mean_pct", "far_holes_moving_median_pct",
+               "arrival_lod02_settled_s")
+    medians = {condition: {metric: statistics.median(
+        row[metric] for row in rows if row["condition"] == condition
+        and row.get(metric) is not None) for metric in metrics}
+        for condition in ("baseline", "candidate")}
+    by_pair = {pair: {row["condition"]: row for row in rows if row["run"] == f"pair-{pair}"}
+               for pair in ORDER}
+    deltas = [
+        {"pair": pair, **{f"{metric}_delta":
+            by_pair[pair]["candidate"][metric] - by_pair[pair]["baseline"][metric]
+            for metric in metrics}}
+        for pair in ORDER
+    ]
+    summary = {"protocol": "Seven interleaved 30 m/s UE pairs using compiled defaults; "
+               "candidate defaults enable the two arrival-gated fast-travel switches; no "
+               "command-line switch overrides; 12 s move plus 5 s dwell; eight effective "
+               "tile slots; one warm-up per binary excluded.",
+               "defaults": {"incremental_desired_set": 1,
+                            "streaming_fast_travel_wait_for_all_critical_tiles": 0,
+                            "streaming_fast_travel_apply_budget_ms": 1.0,
+                            "forward_first": 0, "rebuild_on_core_region_completion": 1,
+                            "core_desired_cache_by_center": 0,
+                            "incremental_desired_audit_every": 0,
+                            "incremental_desired_materialization_audit_every": 0,
+                            "incremental_desired_audit_inject_mismatch": 0},
+               "warmups": warmups, "movement_medians": medians,
+               "paired_deltas_candidate_minus_baseline": deltas, "runs": rows}
+    (OUT / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n",
+                                       encoding="utf-8")
+    print("[L170Summary] " + json.dumps({"movement_medians": medians,
+                                         "paired_deltas": deltas}, sort_keys=True), flush=True)
+    all_runs_p99_pass = all(row["tick_p99_ms"] is not None and row["tick_p99_ms"] <= 11.0
+                            for row in rows)
+    all_runs_worst_pass = all(row["tick_worst_ms"] is not None
+                              and row["tick_worst_ms"] <= 14.0 for row in rows)
+    candidate_worst = [row["tick_worst_ms"] for row in rows
+                       if row["condition"] == "candidate"]
+    baseline_worst = [row["tick_worst_ms"] for row in rows
+                      if row["condition"] == "baseline"]
+    arrival_pairs_not_slower = sum(
+        by_pair[pair]["candidate"]["arrival_lod02_settled_s"]
+        <= by_pair[pair]["baseline"]["arrival_lod02_settled_s"]
+        for pair in ORDER)
+    arrival_median_delta_pct = 100.0 * (
+        medians["candidate"]["arrival_lod02_settled_s"]
+        / medians["baseline"]["arrival_lod02_settled_s"] - 1.0)
+    near_gain_pp = (medians["candidate"]["near_readiness_moving_mean_pct"]
+                    - medians["baseline"]["near_readiness_moving_mean_pct"])
+    forward_gain_pp = (medians["candidate"]["forward_cone_readiness_moving_mean_pct"]
+                       - medians["baseline"]["forward_cone_readiness_moving_mean_pct"])
+    far_hole_reduction_pp = (medians["baseline"]["far_holes_moving_median_pct"]
+                             - medians["candidate"]["far_holes_moving_median_pct"])
+    gate_metrics = {
+        "arrival_delta_pct": arrival_median_delta_pct,
+        "arrival_pairs_not_slower": arrival_pairs_not_slower,
+        "near_gain_pp": near_gain_pp,
+        "forward_gain_pp": forward_gain_pp,
+        "far_hole_reduction_pp": far_hole_reduction_pp,
+    }
+    gates = {
+        "every_run_p99_le_11_ms": all_runs_p99_pass,
+        "every_run_worst_le_14_ms": all_runs_worst_pass,
+        "candidate_p99_median_not_worse": medians["candidate"]["tick_p99_ms"]
+            <= medians["baseline"]["tick_p99_ms"],
+        "candidate_worst_tick_median_not_worse": bool(candidate_worst and baseline_worst)
+            and statistics.median(candidate_worst) <= statistics.median(baseline_worst),
+        "near_gain_at_least_75pct_of_l169": near_gain_pp >= 9.63,
+        "forward_gain_at_least_75pct_of_l169": forward_gain_pp >= 1.81,
+        "far_hole_reduction_at_least_75pct_of_l169": far_hole_reduction_pp >= 1.35,
+        "arrival_median_within_0_5pct": abs(arrival_median_delta_pct) <= 0.5,
+        "arrival_not_slower_in_majority_of_pairs": arrival_pairs_not_slower >= 4,
+    }
+    summary["promotion_gates"] = gates
+    summary["promotion_gate_metrics"] = gate_metrics
+    (OUT / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n",
+                                       encoding="utf-8")
+    print("[L170Gates] " + json.dumps({"checks": gates, "metrics": gate_metrics},
+                                      sort_keys=True), flush=True)
+    return 0 if all(gates.values()) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())