streaming: fast-travel near readiness in Unreal (L108)

Policy 3 movement traces (12 s move + 5 s dwell, 3 runs each). Unreal 30 m/s: near LOD0-2
ready while moving 31.6 -> 44.2% (LOD0 15.4 -> 34.7%), world tick p99/max 23.1/28.3 ->
11.3/13.6 ms, dwell completion 3.55 -> 3.22 s; far holes 2.5 -> 4.3%. Walking unchanged (~99%).
Stale work is small (1.9% of moving LOD0 tiles finish out of range); the remaining limit is
sustained generation throughput (8 task slots, p95 worker 105 ms). Near:far stays 4:1.
World unchanged (pinned hashes pass). Fast tier PASS; Linux UE builds.

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

Standalone/Research/WorldAssemblerM3/Linux108/REPORT.md

@@ -0,0 +1,31 @@
+# Linux108 movement results
+
+Baseline is `ef22b03`. Policy 3; three runs per speed and platform; 12 s moving plus 5 s dwell. Percentages and dwell times are the median with the three-run range. UE tick is `commandlet_world_tick_frame.world_tick_ms`; headless records sampled queue-update work, not frame ticks. Tick/update values are medians of the per-run p99 and max. Columns show baseline → current.
+
+| Speed / platform | Near LOD0–2 | Walkable ring | Far holes | Tick or update p99 / max (ms) | Dwell complete (s) |
+|---|---:|---:|---:|---:|---:|
+| 1.4 m/s / headless | 89.40 [89.23–89.43] → 89.57 [89.43–89.85] | 97.92 → 97.92 | 13.72 [13.59–14.11] → 13.46 [12.98–13.69] | 4.37 / 4.37 → 4.44 / 4.44 | 0.00 → 0.00 |
+| 1.4 m/s / UE | 99.27 [99.21–99.28] → 99.31 [99.28–99.43] | 100 → 100 | 0.62 [0.56–0.85] → 1.26 [1.18–1.36] | 11.48 / 26.39 → 6.25 / 14.28 | 0.07 [0.04–0.18] → 0.02 [0.02–0.24] |
+| 30 m/s / headless | 80.64 [80.60–81.10] → 82.61 [82.41–82.77] | 95.14 → 96.53 [96.30–96.53] | 6.55 [6.35–6.64] → 14.42 [14.14–14.66] | 4.22 / 4.22 → 4.35 / 4.35 | 0.25 → 0.25 |
+| 30 m/s / UE | 31.55 [29.13–31.83] → 44.24 [43.85–45.70] | 100 → 100 | 2.52 [2.14–3.15] → 4.25 [4.03–4.66] | 23.11 / 28.29 → 11.29 / 13.60 | 3.55 [3.44–3.61] → 3.22 [3.18–3.29] |
+
+UE per-LOD readiness, baseline → current (median [range]):
+
+| Speed | LOD0 | LOD1 | LOD2 |
+|---|---:|---:|---:|
+| 1.4 m/s | 99.02 [99.00–99.06] → 99.11 [99.06–99.14] | 99.79 [99.77–99.82] → 99.84 [99.82–99.84] | 98.90 [98.83–99.04] → 98.92 [98.90–99.38] |
+| 30 m/s | 15.42 [13.16–15.49] → 34.68 [34.50–35.94] | 25.62 [23.29–26.44] → 42.06 [41.44–43.91] | 59.07 [57.33–61.11] → 59.60 [59.20–60.88] |
+
+## What the traces show
+
+At 30 m/s, the scheduler changes desired center 45 times per run, about once per 8 m tile crossing. 88.4% of moving LOD0 admissions are ahead of the heading; median lead is 13.2 m (p95 27.6 m). Of 4,835 generated LOD0 tiles admitted during movement, 94 (1.94%) finish outside the desired range. 29 (0.60%) are cancelled before worker completion. Stale work is not the main readiness loss.
+
+LOD0 readiness starts at 83–85% in the first two seconds and falls to 14–16% in the final two. About 512–516 LOD0 tiles remain desired; the trace shows a mean of about 328 missing. The worker task cap is 8, with p95 worker time 105 ms and p95 request-to-apply time 155 ms. This points to sustained generation/backlog throughput as the remaining fast-travel limit. The 2 ms apply budget is rarely hit (4–7 frames per run); tile-apply max is under 2.5 ms. `UWorld::Tick` max is 14.7 ms; the custom streaming-work tick max is 13.7 ms. The commandlet loop includes a deliberate ~16 ms sleep and is excluded from these tick values.
+
+Fast near readiness improved by 12.7 points, but remains 44.2% in UE versus 82.6% headless. Near-first service also costs far coverage: 30 m/s far holes rose from 2.52% to 4.25% in UE and 6.55% to 14.42% headless. A headless 2:1 near/far trial measured 80.1% near and 14.8% far holes, so the final scheduler retains 4:1. Walk/fast hysteresis remains 75/35 and 1200/800 cm/s.
+
+## Checks
+
+The Unreal build succeeded with `-NoUBA -MaxParallelActions=4` before timing. Unreal timings used `taskset` only; headless used `taskset` plus the 6 GiB `prlimit`. The final headless build stamp is `f3ec37a389bb6c984bac0fa360b96e49b6b53fbda92c6461c1246b690e21afaf`, matching the fast-tier manifest. Fast tier passed: 274 checks, 80 runs; pinned L100 hashes and W1/W8/W12 comparisons passed. `git diff --check` is clean. No passage-support or mesher geometry semantics changed.
+
+Artifacts: `Saved/Linux/l108/movement/{baseline,after}/summary.json`, `after/tile-timelines.csv`, Unreal traces, and `Saved/Linux/l108/verify-fast/fast-summary.json`.

Standalone/Research/WorldAssemblerM3/Linux108/RunL108Movement.py

@@ -0,0 +1,475 @@
+#!/usr/bin/env python3
+"""Three-run L80 movement protocol for headless and Unreal, policy 3 only."""
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import math
+import re
+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")
+DATA = ROOT / "Standalone/data/owner_world.json"
+HEADLESS = {
+    "baseline": ROOT / "Saved/Linux/l108/bin/VoxelForgeCoreHeadless-l108-baseline-clang",
+    "after": ROOT / "Saved/Linux/l108/bin/VoxelForgeCoreHeadless-l108-after-clang",
+}
+OUT = ROOT / "Saved/Linux/l108/movement"
+HOST_OUT = HOST / "Plugins/VoxelForge/Saved/Linux/l108/movement"
+SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375)
+SPEEDS = {"walk": 1.4, "fast": 30.0}
+SPEED_CM_S = {"walk": 140.0, "fast": 3000.0}
+TRACE_EVENT = re.compile(r"^\{.*\}$")
+HEADLESS_SAMPLE = re.compile(r"^\[L79MovingSample\] (.*)$", re.M)
+HEADLESS_SUMMARY = re.compile(r"^\[L79MovingSummary\] (.*)$", re.M)
+FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
+
+
+def p99(values: list[float]) -> float | None:
+    if not values:
+        return None
+    values.sort()
+    return values[max(0, math.ceil(0.99 * len(values)) - 1)]
+
+
+def mean(values: list[float]) -> float | None:
+    return sum(values) / len(values) if values else None
+
+
+def parse_events(path: Path) -> list[dict]:
+    events = []
+    if not path.is_file():
+        return events
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        if not TRACE_EVENT.match(line):
+            continue
+        try:
+            events.append(json.loads(line))
+        except json.JSONDecodeError:
+            pass
+    return events
+
+
+def summarize_rows(rows: list[dict], moving_key: str = "move") -> dict:
+    moving = [r for r in rows if r["phase"] == moving_key]
+    dwell = [r for r in rows if r["phase"] == "dwell"]
+    dwell_end = dwell[-1] if dwell else {}
+    full_dwell = next((r for r in dwell
+                       if r["near_ready_pct"] >= 99.999
+                       and r["far_hole_pct"] <= 0.001), None)
+    return {
+        "moving_samples": len(moving),
+        "dwell_samples": len(dwell),
+        "near_ready_moving_mean_pct": mean([r["near_ready_pct"] for r in moving]),
+        "lod0_ready_moving_mean_pct": mean([r["lod0_ready_pct"] for r in moving if "lod0_ready_pct" in r]),
+        "lod1_ready_moving_mean_pct": mean([r["lod1_ready_pct"] for r in moving if "lod1_ready_pct" in r]),
+        "lod2_ready_moving_mean_pct": mean([r["lod2_ready_pct"] for r in moving if "lod2_ready_pct" in r]),
+        "walkable_ring_moving_mean_pct": mean([r["walkable_ring_ready_pct"] for r in moving]),
+        "far_holes_moving_mean_pct": mean([r["far_hole_pct"] for r in moving]),
+        "update_work_p99_ms": p99([float(r["queue_update_ms"]) for r in moving
+                                    if "queue_update_ms" in r]),
+        "update_work_max_ms": max([float(r["queue_update_ms"]) for r in moving
+                                   if "queue_update_ms" in r], default=None),
+        "near_ready_dwell_end_pct": dwell_end.get("near_ready_pct"),
+        "walkable_ring_dwell_end_pct": dwell_end.get("walkable_ring_ready_pct"),
+        "far_holes_dwell_end_pct": dwell_end.get("far_hole_pct"),
+        "dwell_complete_after_stop_s": full_dwell.get("since_stop_s") if full_dwell else None,
+    }
+
+
+def run_headless(phase: str, speed_name: str, repetition: int) -> dict:
+    label = f"{speed_name}-r{repetition}"
+    folder = OUT / phase / "headless"
+    folder.mkdir(parents=True, exist_ok=True)
+    log = folder / f"{label}.log"
+    camera = ",".join(f"{v:.12f}" for v in SPAWN_M) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "timeout", "1740s", str(HEADLESS[phase]), str(DATA),
+        "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
+        "--workers=8", "--clip-radius=3", "--max-clip-level=8", "--max-level=8",
+        f"--camera={camera}", "--profile-lod-work", "--stream-scheduler-policy=3",
+        f"--moving-camera-speed-mps={SPEEDS[speed_name]:.1f}",
+        "--moving-camera-duration-s=12", "--moving-camera-dwell-s=5",
+    ]
+    started = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    content = log.read_text(encoding="utf-8", errors="replace")
+    rows = []
+    for match in HEADLESS_SAMPLE.finditer(content):
+        row = dict(FIELDS.findall(match.group(1)))
+        row.update({key: float(row[key]) for key in (
+            "t_s", "near_ready_pct", "walkable_ring_ready_pct", "far_hole_pct",
+            "queue_update_ms")})
+        row["phase"] = row["phase"]
+        row["walkable_ring_ready_pct"] = float(row.get("walkable_ring_ready_pct", 0.0))
+        row["since_stop_s"] = max(0.0, float(row["t_s"]) - 12.0)
+        rows.append(row)
+    summary_match = HEADLESS_SUMMARY.search(content)
+    engine_summary = dict(FIELDS.findall(summary_match.group(1))) if summary_match else {}
+    if result.returncode or not rows or int(engine_summary.get("generation_failures", "-1")) != 0:
+        raise RuntimeError(f"headless movement run failed: {log}; exit={result.returncode}")
+    info = {
+        "phase": phase, "platform": "headless", "speed": speed_name,
+        "speed_mps": SPEEDS[speed_name], "repetition": repetition,
+        "exit_code": result.returncode, "wall_s": round(time.monotonic() - started, 3),
+        "log": str(log.relative_to(ROOT)), **summarize_rows(rows),
+    }
+    return info
+
+
+def run_unreal(phase: str, speed_name: str, repetition: int) -> tuple[dict, list[dict], list[dict]]:
+    label = f"{speed_name}-r{repetition}"
+    folder = OUT / phase / "unreal"
+    host_folder = HOST_OUT / phase / "unreal"
+    folder.mkdir(parents=True, exist_ok=True)
+    trace = folder / f"{label}.trace.jsonl"
+    abslog = folder / f"{label}.log"
+    stdout = folder / f"{label}.stdout.log"
+    export = HOST / "Plugins/VoxelForge/Saved/Linux/l108/output" / f"{phase}-{label}"
+    userdir = HOST / "Saved/Linux/l108/user" / f"{phase}-{label}"
+    zendata = HOST / "Saved/Linux/l108/zen" / f"{phase}-{label}"
+    ddc = HOST / "Saved/Linux/l108/DDC"
+    for path in (export, userdir, zendata, ddc):
+        path.mkdir(parents=True, exist_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.CoreMCGradientMemoization=1",
+        "-voxel.StreamingCollisionLookahead=1", "-voxel.TestMove=1",
+        f"-voxel.TestMoveSpeedCmPerSecond={SPEED_CM_S[speed_name]:.1f}",
+        "-voxel.TestMoveStartSeconds=6.0", "-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}",
+    ]
+    wrapped = ["taskset", "-c", "0-11", "timeout", "1740s", *command]
+    started = time.monotonic()
+    with stdout.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(wrapped, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    events = parse_events(trace)
+    samples = [event for event in events if event.get("type") == "streaming_sample"]
+    move_samples = [event for event in samples if float(event.get("speed_cm_s", 0.0)) >= 50.0]
+    move_start_t = min((float(event.get("t_s", 0.0)) for event in move_samples), default=float("inf"))
+    stop = next((event for event in events
+                 if event.get("type") == "streaming_test_move_stopped"), None)
+    stop_t = float(stop.get("t_s", 0.0)) if stop else float("inf")
+    dwell_samples = [event for event in samples
+                     if float(event.get("t_s", 0.0)) >= stop_t
+                     and float(event.get("speed_cm_s", 0.0)) < 50.0]
+    rows = []
+    for event in samples:
+        desired = event.get("desired_by_lod", [0] * 9)
+        resident = event.get("resident_by_lod", [0] * 9)
+        covered = event.get("covered_by_lod", resident)
+        desired_near = sum(desired[:3])
+        resident_near = sum(resident[:3])
+        desired_far = sum(desired[3:])
+        covered_far = sum(covered[3:])
+        event_time = float(event.get("t_s", 0.0))
+        speed = float(event.get("speed_cm_s", 0.0))
+        row_phase = "move" if speed >= 50.0 else (
+            "dwell" if event_time >= stop_t else "arrival")
+        ring_pct = 100.0 * float(event.get("walkable_ready_tiles", 0)) / max(
+            1.0, float(event.get("walkable_required_tiles", 9)))
+        rows.append({
+            "t_s": event_time, "speed_cm_s": speed, "phase": row_phase,
+            "since_stop_s": max(0.0, event_time - stop_t) if row_phase == "dwell" else None,
+            "near_ready_pct": 100.0 * resident_near / max(1, desired_near),
+            "lod0_ready_pct": 100.0 * resident[0] / max(1, desired[0]),
+            "lod1_ready_pct": 100.0 * resident[1] / max(1, desired[1]),
+            "lod2_ready_pct": 100.0 * resident[2] / max(1, desired[2]),
+            "walkable_ring_ready_pct": ring_pct,
+            "far_hole_pct": 100.0 * (desired_far - covered_far) / max(1, desired_far),
+            "near_desired": desired_near, "near_resident": resident_near,
+            "walkable_ready_tiles": event.get("walkable_ready_tiles"),
+            "epoch": event.get("epoch"),
+        })
+    if len(move_samples) < 40 or len(dwell_samples) < 16:
+        raise RuntimeError(f"incomplete Unreal movement trace: {trace}; "
+                           f"move={len(move_samples)} dwell={len(dwell_samples)} "
+                           f"exit={result.returncode}")
+    frame_samples = [event for event in events if event.get("type") == "apply_frame"
+                     and move_start_t <= float(event.get("t_s", 0.0)) < stop_t]
+    tick_samples = [event for event in events if event.get("type") == "game_thread_work_frame"
+                    and move_start_t <= float(event.get("t_s", 0.0)) < stop_t]
+    tick_ms = [float(event.get("tick_ms", 0.0)) for event in tick_samples]
+    update_ms = [float(event.get("update_chunks_ms", 0.0)) for event in tick_samples]
+    drain_ms = [float(event.get("pending_drain_ms", 0.0)) for event in tick_samples]
+    apply_ms = [float(event.get("apply_ms", 0.0)) for event in frame_samples]
+    gate_epochs = [event for event in events if event.get("type") == "walkable_gate"
+                   and move_start_t <= float(event.get("t_s", 0.0)) < stop_t]
+    summary = summarize_rows(rows)
+    summary.update({
+        "game_thread_tick_p99_ms": p99(tick_ms), "game_thread_tick_max_ms": max(tick_ms, default=None),
+        "stream_update_p99_ms": p99(update_ms), "stream_update_max_ms": max(update_ms, default=None),
+        "pending_drain_p99_ms": p99(drain_ms), "pending_drain_max_ms": max(drain_ms, default=None),
+        "tile_apply_p99_ms": p99(apply_ms), "tile_apply_max_ms": max(apply_ms, default=None),
+        "apply_budget_ms": max((float(event.get("apply_budget_ms", 0.0))
+                                 for event in frame_samples), default=None),
+        "apply_budget_hit_frames": sum(int(event.get("time_limit_reached", 0))
+                                        for event in frame_samples),
+        "desired_epochs_during_move": len({event.get("desired_epoch") for event in gate_epochs}),
+        "desired_centers_during_move": len({tuple(event.get("support_center", [])) for event in gate_epochs}),
+    })
+    info = {
+        "phase": phase, "platform": "unreal", "speed": speed_name,
+        "speed_mps": SPEEDS[speed_name], "repetition": repetition,
+        "exit_code": result.returncode, "wall_s": round(time.monotonic() - started, 3),
+        "moving_samples": len(move_samples), "dwell_samples": len(dwell_samples),
+        "trace": str(trace.relative_to(ROOT)), "stdout": str(stdout.relative_to(ROOT)),
+        **summary,
+    }
+    return info, events, rows
+
+
+def write_tile_timelines(phase: str, unreal_runs: list[tuple[dict, list[dict]]]) -> dict:
+    path = OUT / phase / "tile-timelines.csv"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    columns = ["speed", "repetition", "ordinal", "tile", "level", "priority", "epoch",
+               "desired_epoch", "in_move_window",
+               "request_s", "cancel_s", "worker_end_s", "generated", "aborted", "obsolete",
+               "applied_s", "request_to_apply_s", "worker_queue_s", "worker_task_s",
+               "result_queue_s", "discard_s", "discard_reason", "ahead_m_at_request",
+               "request_to_worker_end_s", "cancel_before_worker_end",
+               "out_of_range_at_worker_end"]
+    stale_by_run = {}
+    with path.open("w", newline="", encoding="utf-8") as stream:
+        writer = csv.DictWriter(stream, fieldnames=columns)
+        writer.writeheader()
+        for run, events in unreal_runs:
+            requests, cancelled, finished, applied, discarded = {}, {}, {}, {}, {}
+            samples = sorted((event for event in events if event.get("type") == "streaming_sample"),
+                             key=lambda event: float(event.get("t_s", 0.0)))
+            move_samples = [event for event in samples
+                            if float(event.get("speed_cm_s", 0.0)) >= 50.0]
+            move_start_s = min((float(event.get("t_s", 0.0)) for event in move_samples),
+                               default=float("inf"))
+            move_stop = next((event for event in events
+                              if event.get("type") == "streaming_test_move_stopped"), None)
+            move_stop_s = float(move_stop.get("t_s", float("inf"))) if move_stop else float("inf")
+
+            def pawn_x_at(t_s: float) -> float | None:
+                if not samples:
+                    return None
+                before = None
+                for sample in samples:
+                    sample_t = float(sample.get("t_s", 0.0))
+                    if sample_t >= t_s:
+                        if before is None:
+                            return float(sample.get("position_cm", [0.0])[0]) / 100.0
+                        t0 = float(before.get("t_s", 0.0))
+                        t1 = sample_t
+                        x0 = float(before.get("position_cm", [0.0])[0]) / 100.0
+                        x1 = float(sample.get("position_cm", [0.0])[0]) / 100.0
+                        alpha = 0.0 if t1 <= t0 else (t_s - t0) / (t1 - t0)
+                        return x0 + alpha * (x1 - x0)
+                    before = sample
+                return float(samples[-1].get("position_cm", [0.0])[0]) / 100.0
+
+            for event in events:
+                ordinal = event.get("request_ordinal")
+                if ordinal is None:
+                    continue
+                ordinal = int(ordinal)
+                kind = event.get("type")
+                if kind == "request":
+                    requests[ordinal] = event
+                elif kind == "tile_cancelled":
+                    cancelled[ordinal] = event
+                elif kind == "tile_worker_end":
+                    finished[ordinal] = event
+                elif kind == "tile":
+                    applied[ordinal] = event
+                elif kind == "tile_discard":
+                    discarded[ordinal] = event
+
+            generated_l0 = 0
+            stale_l0 = 0
+            out_of_range_l0 = 0
+            l0_cancelled_while_running = 0
+            ahead_distances = []
+            gate_epochs = sorted((event for event in events if event.get("type") == "walkable_gate"),
+                                 key=lambda event: float(event.get("t_s", 0.0)))
+
+            def desired_epoch_at(t_s: float) -> int | None:
+                previous = [event for event in gate_epochs if float(event.get("t_s", 0.0)) <= t_s]
+                return int(previous[-1]["desired_epoch"]) if previous else None
+
+            for ordinal, request in requests.items():
+                key = request.get("tile", [None, None, None])
+                level = int(request.get("level", -1))
+                cancel = cancelled.get(ordinal)
+                end = finished.get(ordinal)
+                tile = applied.get(ordinal)
+                discard = discarded.get(ordinal)
+                request_t = float(request.get("t_s", 0.0))
+                end_t = float(end.get("t_s", 0.0)) if end else None
+                cancel_t = float(cancel.get("t_s", 0.0)) if cancel else None
+                # LOD0 clip tiles are 8 m across; the live x-range is [center-4, center+3].
+                if end and int(end.get("generated", 0)) == 1 and end_t is not None:
+                    finish_x = pawn_x_at(end_t)
+                    center_x = math.floor(finish_x / 8.0) if finish_x is not None else 0
+                    physically_out = bool(level == 0 and finish_x is not None and key[0] is not None
+                                          and (int(key[0]) < center_x - 4
+                                               or int(key[0]) > center_x + 3))
+                else:
+                    physically_out = False
+                in_move_window = bool(end_t is not None and move_start_s <= end_t < move_stop_s)
+                if level == 0 and end and int(end.get("generated", 0)) == 1 and in_move_window:
+                    generated_l0 += 1
+                    stale_l0 += int(cancel_t is not None and cancel_t < end_t)
+                    out_of_range_l0 += int(physically_out)
+                if level == 0 and end and int(end.get("aborted", 0)) == 1 \
+                        and cancel_t is not None and end_t is not None and cancel_t < end_t:
+                    l0_cancelled_while_running += 1
+                pawn_x = pawn_x_at(request_t)
+                extent_m = 8.0 * (1 << max(0, level))
+                tile_center_m = (float(key[0]) + 0.5) * extent_m if key[0] is not None else None
+                ahead_m = (tile_center_m - pawn_x) if tile_center_m is not None and pawn_x is not None else None
+                if level == 0 and move_start_s <= request_t < move_stop_s and ahead_m is not None:
+                    ahead_distances.append(ahead_m)
+                writer.writerow({
+                    "speed": run["speed"], "repetition": run["repetition"],
+                    "ordinal": ordinal, "tile": ",".join(str(v) for v in key),
+                    "level": level, "priority": request.get("priority"),
+                    "epoch": request.get("epoch"),
+                    "desired_epoch": request.get("desired_epoch", desired_epoch_at(request_t)),
+                    "in_move_window": int(move_start_s <= request_t < move_stop_s),
+                    "request_s": request_t,
+                    "cancel_s": cancel_t, "worker_end_s": end_t,
+                    "generated": end.get("generated") if end else None,
+                    "aborted": end.get("aborted") if end else None,
+                    "obsolete": end.get("obsolete") if end else None,
+                    "applied_s": float(tile.get("t_s", 0.0)) if tile else None,
+                    "request_to_apply_s": tile.get("request_to_apply_s") if tile else None,
+                    "worker_queue_s": tile.get("worker_queue_s") if tile else None,
+                    "worker_task_s": tile.get("worker_task_s") if tile else (
+                        end.get("worker_task_s") if end else None),
+                    "result_queue_s": tile.get("result_queue_s") if tile else None,
+                    "discard_s": float(discard.get("t_s", 0.0)) if discard else None,
+                    "discard_reason": discard.get("reason") if discard else None,
+                    "ahead_m_at_request": ahead_m,
+                    "request_to_worker_end_s": end_t - request_t if end_t is not None else None,
+                    "cancel_before_worker_end": int(bool(cancel_t is not None
+                                                         and end_t is not None
+                                                         and cancel_t < end_t)),
+                    "out_of_range_at_worker_end": int(physically_out),
+                })
+            key = f"{run['speed']}-r{run['repetition']}"
+            stale_by_run[key] = {
+                "generated_lod0_tiles_finished_during_move": generated_l0,
+                "generated_lod0_out_of_range_at_finish": out_of_range_l0,
+                "out_of_range_fraction_pct": 100.0 * out_of_range_l0 / max(1, generated_l0),
+                "generated_lod0_finished_after_cancel": stale_l0,
+                "cancel_before_finish_fraction_pct": 100.0 * stale_l0 / max(1, generated_l0),
+                "lod0_cancelled_while_worker_active": l0_cancelled_while_running,
+                "lod0_admissions_in_move": len(ahead_distances),
+                "lod0_ahead_at_admission_pct": 100.0 * sum(value > 0.0 for value in ahead_distances)
+                    / max(1, len(ahead_distances)),
+                "lod0_ahead_median_m": statistics.median(ahead_distances) if ahead_distances else None,
+            }
+    return {"tile_timelines_csv": str(path.relative_to(ROOT)), "stale_by_run": stale_by_run}
+
+
+def run_phase(phase: str) -> None:
+    partial_path = OUT / phase / "runs-partial.json"
+    results = []
+    if partial_path.is_file():
+        try:
+            stored = json.loads(partial_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            stored = []
+        for item in stored:
+            if item.get("phase") != phase or int(item.get("exit_code", -1)) != 0:
+                continue
+            if item.get("platform") == "unreal":
+                trace = ROOT / item.get("trace", "")
+                if not trace.is_file() or not Path(str(trace) + ".done").is_file():
+                    continue
+            elif item.get("platform") == "headless":
+                log = ROOT / item.get("log", "")
+                if not log.is_file() or "[L79MovingSummary]" not in log.read_text(
+                        encoding="utf-8", errors="replace"):
+                    continue
+                content = log.read_text(encoding="utf-8", errors="replace")
+                update_ms = []
+                for match in HEADLESS_SAMPLE.finditer(content):
+                    row = dict(FIELDS.findall(match.group(1)))
+                    if row.get("phase") == "move" and "queue_update_ms" in row:
+                        update_ms.append(float(row["queue_update_ms"]))
+                item["update_work_p99_ms"] = p99(update_ms)
+                item["update_work_max_ms"] = max(update_ms, default=None)
+            results.append(item)
+    completed = {(item["platform"], item["speed"], item["repetition"])
+                 for item in results}
+    unreal_runs = []
+    for item in results:
+        if item["platform"] == "unreal":
+            trace = ROOT / item["trace"]
+            unreal_runs.append((item, parse_events(trace)))
+
+    def save_partial() -> None:
+        partial_path.parent.mkdir(parents=True, exist_ok=True)
+        partial_path.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8")
+
+    for repetition in range(1, 4):
+        speed_order = ("walk", "fast") if repetition % 2 else ("fast", "walk")
+        for speed_name in speed_order:
+            headless_key = ("headless", speed_name, repetition)
+            if headless_key not in completed:
+                headless_info = run_headless(phase, speed_name, repetition)
+                results.append(headless_info)
+                completed.add(headless_key)
+                save_partial()
+                print("[L108MovementRun] " + json.dumps(headless_info, sort_keys=True), flush=True)
+            unreal_key = ("unreal", speed_name, repetition)
+            if unreal_key not in completed:
+                run_info, events, _rows = run_unreal(phase, speed_name, repetition)
+                results.append(run_info)
+                unreal_runs.append((run_info, events))
+                completed.add(unreal_key)
+                save_partial()
+                print("[L108MovementRun] " + json.dumps(run_info, sort_keys=True), flush=True)
+    timeline = write_tile_timelines(phase, unreal_runs)
+    (OUT / phase / "summary.json").write_text(
+        json.dumps({"phase": phase, "runs": results, "timeline": timeline}, indent=2) + "\n",
+        encoding="utf-8")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("phase", choices=("baseline", "after"))
+    args = parser.parse_args()
+    if not DATA.is_file() or not HEADLESS[args.phase].is_file() or not EDITOR.is_file():
+        raise SystemExit("missing movement runner input, headless executable or Unreal editor")
+    run_phase(args.phase)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

Standalone/Source/StreamBenchmarkPipeline.h

@@ -5631,6 +5631,7 @@ namespace StandaloneStreamBenchmark
                     return 9;
                 }
                 std::size_t NearDesired=0,NearResident=0,FarDesired=0,FarResident=0,FarCovered=0;
+                std::size_t WalkableRingResident=0;
                 double FarStaleAgeSum=0.0,FarStaleAgeMax=0.0;
                 const auto FloorDiv=[](int32 Value,int32 Divisor)
                 {
@@ -5672,6 +5673,19 @@ namespace StandaloneStreamBenchmark
                         }
                     }
                 }
+                for(int32 Dy=-1;Dy<=1;++Dy)
+                for(int32 Dx=-1;Dx<=1;++Dx)
+                {
+                    const FVoxelTileKey WalkableKey(
+                        FIntVector(CurrentDesired.CenterChunk.X+Dx,
+                                   CurrentDesired.CenterChunk.Y+Dy,
+                                   CurrentDesired.CenterChunk.Z),0);
+                    const bool bWalkableKeyDesired=std::find(
+                        CurrentDesired.Tiles.begin(),CurrentDesired.Tiles.end(),WalkableKey)
+                        !=CurrentDesired.Tiles.end();
+                    if(bWalkableKeyDesired&&Resident.contains(WalkableKey))
+                        ++WalkableRingResident;
+                }
                 FarMissingSince.swap(CurrentFarMissing);
                 const double NearReadyPercent=100.0*static_cast<double>(NearResident)
                     /static_cast<double>(std::max<std::size_t>(1,NearDesired));
@@ -5681,12 +5695,15 @@ namespace StandaloneStreamBenchmark
                     /static_cast<double>(std::max<std::size_t>(1,FarDesired));
                 const double FarStaleAgeMean=FarDesired>FarResident
                     ?FarStaleAgeSum/static_cast<double>(FarDesired-FarResident):0.0;
+                const double WalkableRingReadyPercent=100.0
+                    *static_cast<double>(WalkableRingResident)/9.0;
                 if(FarDesired>0&&FarDesired==FarResident&&FirstAllFarReadySeconds<0.0)
                     FirstAllFarReadySeconds=ElapsedSeconds;
                 const char* Phase=ElapsedSeconds<MovingCameraDurationSeconds?"move":"dwell";
-                std::printf("[L79MovingSample] policy=%d speed_mps=%.3f t_s=%.3f phase=%s x_m=%.3f near=%zu/%zu near_ready_pct=%.3f far=%zu/%zu far_missing_pct=%.3f far_covered=%zu/%zu far_hole_pct=%.3f far_stale_mean_s=%.3f far_stale_max_s=%.3f queued_jobs=%d queue_update_ms=%.3f generation_failures=%d\n",
+                std::printf("[L79MovingSample] policy=%d speed_mps=%.3f t_s=%.3f phase=%s x_m=%.3f near=%zu/%zu near_ready_pct=%.3f walkable_ring=%zu/9 walkable_ring_ready_pct=%.3f far=%zu/%zu far_missing_pct=%.3f far_covered=%zu/%zu far_hole_pct=%.3f far_stale_mean_s=%.3f far_stale_max_s=%.3f queued_jobs=%d queue_update_ms=%.3f generation_failures=%d\n",
                     SchedulerPolicy,MovingCameraSpeedMetersPerSecond,ElapsedSeconds,Phase,
                     Camera.X*VoxelMetres,NearResident,NearDesired,NearReadyPercent,
+                    WalkableRingResident,WalkableRingReadyPercent,
                     FarResident,FarDesired,FarMissingPercent,FarCovered,FarDesired,
                     FarHolePercent,FarStaleAgeMean,FarStaleAgeMax,
                     MovingUpdate.NewJobs,UpdateMilliseconds,GenerationFailures);

Standalone/Source/StreamingWorld.h

@@ -996,6 +996,7 @@ private:
                 else if(Key.Level<=2)NearTiles.push_back(Key);
                 else FarTiles.push_back(Key);
             }
+            const int32 SpeedAwareMode=SpeedAwareSchedulerMode.load(std::memory_order_relaxed);
             std::unordered_map<FVoxelTileKey,std::size_t,FStandaloneTileKeyHash>
                 PassageScores;
             if(!PassagePrioritySegments.empty())
@@ -1026,6 +1027,40 @@ private:
                         const std::size_t ScoreB=PassageScores.at(B);
                         if(ScoreA!=ScoreB)return ScoreA>ScoreB;
                     }
+                    if(SpeedAwareMode==2&&bNearFirst
+                        &&TravelDirection.X*TravelDirection.X
+                            +TravelDirection.Y*TravelDirection.Y>1.0e-12f)
+                    {
+                        const double TileScale=static_cast<double>(1<<A.Level);
+                        const double AlongA=
+                            ((static_cast<double>(A.Coord.X)+0.5)*TileScale
+                                -NewDesired.CenterChunk.X)*TravelDirection.X
+                            +((static_cast<double>(A.Coord.Y)+0.5)*TileScale
+                                -NewDesired.CenterChunk.Y)*TravelDirection.Y;
+                        const double AlongB=
+                            ((static_cast<double>(B.Coord.X)+0.5)*TileScale
+                                -NewDesired.CenterChunk.X)*TravelDirection.X
+                            +((static_cast<double>(B.Coord.Y)+0.5)*TileScale
+                                -NewDesired.CenterChunk.Y)*TravelDirection.Y;
+                        const auto LeadRank=[TileScale](double Projection)
+                        {
+                            if(Projection>=TileScale&&Projection<=TileScale*2.0)return 0;
+                            if(Projection>0.0&&Projection<TileScale)return 1;
+                            if(Projection>TileScale*2.0)return 2;
+                            return 3;
+                        };
+                        const int LeadRankA=LeadRank(AlongA),LeadRankB=LeadRank(AlongB);
+                        if(LeadRankA!=LeadRankB)return LeadRankA<LeadRankB;
+                        if(LeadRankA==0)
+                        {
+                            const double TargetLead=TileScale*1.5;
+                            const double ErrorA=std::abs(AlongA-TargetLead);
+                            const double ErrorB=std::abs(AlongB-TargetLead);
+                            if(ErrorA!=ErrorB)return ErrorA<ErrorB;
+                        }
+                        else if(LeadRankA==1&&AlongA!=AlongB)return AlongA>AlongB;
+                        else if(LeadRankA==2&&AlongA!=AlongB)return AlongA<AlongB;
+                    }
                     const bool bAheadA=IsAhead(A),bAheadB=IsAhead(B);
                     if(bAheadA!=bAheadB)return bAheadA;
                     const double DA=StandaloneWorldMesher::TileDistanceSquaredChunks(A,NewDesired.CenterChunk);
@@ -1042,12 +1077,19 @@ private:
             SpeedAwareOrder.insert(SpeedAwareOrder.end(),WalkableRing.begin(),WalkableRing.end());
             const double SpeedMetersPerSecond=SchedulerMotionSpeedMetersPerSecond.load(
                 std::memory_order_relaxed);
-            const int32 SpeedAwareMode=SpeedAwareSchedulerMode.load(std::memory_order_relaxed);
             if(SpeedAwareMode==2)
             {
-                // Fast travel keeps parent coverage ahead of detail work.
-                SpeedAwareOrder.insert(SpeedAwareOrder.end(),FarTiles.begin(),FarTiles.end());
-                SpeedAwareOrder.insert(SpeedAwareOrder.end(),NearTiles.begin(),NearTiles.end());
+                // Fast travel keeps a near-heavy admission ratio and a bounded horizon lane.
+                size_t NearIndex=0,FarIndex=0;
+                while(NearIndex<NearTiles.size())
+                {
+                    const size_t NearEnd=std::min(NearTiles.size(),NearIndex+4);
+                    SpeedAwareOrder.insert(SpeedAwareOrder.end(),
+                        NearTiles.begin()+NearIndex,NearTiles.begin()+NearEnd);
+                    NearIndex=NearEnd;
+                    if(FarIndex<FarTiles.size())SpeedAwareOrder.push_back(FarTiles[FarIndex++]);
+                }
+                SpeedAwareOrder.insert(SpeedAwareOrder.end(),FarTiles.begin()+FarIndex,FarTiles.end());
             }
             else if(SpeedAwareMode==1)
             {