streaming: one supplemental far apply per frame when the budget is half unused (L121)

After the normal visible-apply count is reached and <50% of the 2 ms budget is spent, apply one horizon
result (off during startup boost; voxel.StreamingFarApplySupplement=0 disables). 30 m/s UE, 5 pairs:
far holes 4.38% -> 2.70%, far finish-to-apply p50 1.105 -> 0.114 s, worst tick 14.74 ms (<15.43),
near coverage +0.3 pt, arrival whole-view -313 ms. Geometry unchanged (24/24 hashes), fast tier green.
Adds trace-only RMC proxy publication timing.

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

Source/VoxelForge/Private/VoxelStartupTrace.cpp

@@ -423,6 +423,7 @@ void RecordApplyFrame(const FApplyFrameSample& Sample)
              "\"active_tasks\":%d,\"running_workers\":%d,\"task_cap\":%d,"
              "\"pending_tiles\":%d,\"pending_results_estimate\":%d,"
              "\"results_drained\":%d,\"apply_results\":%d,\"meshes_applied\":%d,\"max_applies\":%d,"
+             "\"supplemental_far_applies\":%d,"
              "\"apply_budget_ms\":%s,\"count_limit_reached\":%d,\"time_limit_reached\":%d,"
              "\"pending_by_lod\":%s,\"apply_results_by_lod\":%s,"
              "\"meshes_by_lod\":%s,\"apply_ms_by_lod\":%s"),
@@ -432,7 +433,8 @@ void RecordApplyFrame(const FApplyFrameSample& Sample)
         Sample.RunningWorkerTasks, Sample.TaskCap, Sample.PendingTiles,
         Sample.PendingResultsEstimate, Sample.ResultsDrained, Sample.ApplyResultsByLevelTotal,
         Sample.MeshesApplied,
-        Sample.MaxApplies, *Number(Sample.ApplyBudgetMilliseconds),
+        Sample.MaxApplies, Sample.SupplementalFarApplies,
+        *Number(Sample.ApplyBudgetMilliseconds),
         Sample.bCountLimitReached ? 1 : 0, Sample.bTimeLimitReached ? 1 : 0,
         *PendingByLevel, *ApplyResultsByLevel, *MeshesByLevel, *ApplyMillisecondsByLevel));
 }

Source/VoxelForge/Private/VoxelWorld.cpp

@@ -268,6 +268,12 @@ namespace
         GVoxelForgeStreamingResultBacklogCap,
         TEXT("Maximum queued results plus admitted tile tasks; 0 disables the safety cap. This is separate from the worker-slot cap."));
 
+    int32 GVoxelForgeStreamingFarApplySupplement = 1;
+    FAutoConsoleVariableRef CVarVoxelForgeStreamingFarApplySupplement(
+        TEXT("voxel.StreamingFarApplySupplement"),
+        GVoxelForgeStreamingFarApplySupplement,
+        TEXT("After the normal visible-apply count is reached, allow one far mesh apply when half of the frame time budget remains; 0 disables the supplemental far lane."));
+
     int32 GVoxelForgeStreamingReadyQueueDepth = 64;
     FAutoConsoleVariableRef CVarVoxelForgeStreamingReadyQueueDepth(
         TEXT("voxel.StreamingReadyQueueDepth"),
@@ -2223,6 +2229,12 @@ void AVoxelWorld::BeginPlay()
     }
     FParse::Value(FCommandLine::Get(), TEXT("voxel.StreamingStartupApplyBoost="),
                   GVoxelForgeStreamingStartupApplyBoost);
+    if (FParse::Value(FCommandLine::Get(), TEXT("voxel.StreamingFarApplySupplement="),
+                      GVoxelForgeStreamingFarApplySupplement))
+    {
+        GVoxelForgeStreamingFarApplySupplement =
+            GVoxelForgeStreamingFarApplySupplement != 0 ? 1 : 0;
+    }
     // Headless BuildHost does not reliably apply arbitrary console-variable command-line tokens
     // before BeginPlay; parse the test switches explicitly so A/B runs exercise the requested path.
     // BuildHost sans interface n'applique pas toujours les CVar arbitraires avant BeginPlay ;
@@ -2737,6 +2749,7 @@ void AVoxelWorld::BeginPlay()
                  "\"streaming_collision_lookahead\":%d,\"core_region_concurrent_admission\":%d,"
                  "\"collision_gate\":%d,"
                  "\"streaming_result_backlog_cap\":%d,\"streaming_startup_apply_boost\":%d,"
+                 "\"streaming_far_apply_supplement\":%d,"
                  "\"streaming_background_worker_reserve\":%d,\"core_parallel_for_background_priority\":%d,"
                  "\"streaming_tile_background_low\":%d,\"streaming_tile_background_normal\":%d,"
                  "\"core_column_cache_direct_mapped\":%d,"
@@ -2758,6 +2771,7 @@ void AVoxelWorld::BeginPlay()
             GVoxelForgeStreamingCollisionLookahead, GVoxelForgeCoreRegionConcurrentAdmission,
             GVoxelForgeCollisionGate,
             GVoxelForgeStreamingResultBacklogCap, GVoxelForgeStreamingStartupApplyBoost,
+            GVoxelForgeStreamingFarApplySupplement,
             GVoxelForgeStreamingBackgroundWorkerReserve,
             GVoxelForgeCoreParallelForBackgroundPriority,
             GVoxelForgeStreamingTileBackgroundLow,
@@ -9881,6 +9895,7 @@ void AVoxelWorld::ProcessPendingChunks()
     const double ApplyBudgetSeconds =
         static_cast<double>(ApplyBudgetMilliseconds) * 0.001;
     int32 MeshesApplied = 0;
+    int32 SupplementalFarApplies = 0;
     int32 ResultsDrained = 0;
     int32 ApplyResultsByLevelTotal = 0;
     double ApplySeconds = 0.0;
@@ -9888,6 +9903,8 @@ void AVoxelWorld::ProcessPendingChunks()
     TStaticArray<int32, 9> MeshesAppliedByLevel{};
     TStaticArray<double, 9> ApplySecondsByLevel{};
     bool bTimeBudgetHit = false;
+    bool bForceSupplementalFarDequeue = false;
+    bool bDequeuedSupplementalFar = false;
     const auto RecordApplyFrame = [&]()
     {
         if (!VoxelForgeStartupTrace::IsActive()) return;
@@ -9911,6 +9928,7 @@ void AVoxelWorld::ProcessPendingChunks()
         Sample.ResultsDrained = ResultsDrained;
         Sample.ApplyResultsByLevelTotal = ApplyResultsByLevelTotal;
         Sample.MeshesApplied = MeshesApplied;
+        Sample.SupplementalFarApplies = SupplementalFarApplies;
         Sample.MaxApplies = MaxApplies;
         Sample.ApplyBudgetMilliseconds = ApplyBudgetMilliseconds;
         Sample.bCountLimitReached = MeshesApplied >= MaxApplies;
@@ -9939,6 +9957,7 @@ void AVoxelWorld::ProcessPendingChunks()
     FChunkResult DequeuedChunk;
     const auto DequeueNextResult = [&]() -> bool
     {
+        bDequeuedSupplementalFar = false;
         if (bStep1UploadCorpusReleased
             && Step1HeldApplyCursor < Step1HeldApplyResults.Num())
         {
@@ -9954,8 +9973,19 @@ void AVoxelWorld::ProcessPendingChunks()
         // Mirror the admission order. A horizon result that completed earlier must not consume
         // the game-thread apply budget ahead of collision support, an edit, or ordinary LOD0.
         if (CriticalProcessQueue.Dequeue(DequeuedChunk)
-            || EditedProcessQueue.Dequeue(DequeuedChunk)
+            || EditedProcessQueue.Dequeue(DequeuedChunk))
-            || NearProcessQueue.Dequeue(DequeuedChunk)
+        {
+            RecordStreamingResultDequeued(DequeuedChunk);
+            return true;
+        }
+        if (bForceSupplementalFarDequeue)
+        {
+            if (!ProcessQueue.Dequeue(DequeuedChunk)) return false;
+            bDequeuedSupplementalFar = true;
+            RecordStreamingResultDequeued(DequeuedChunk);
+            return true;
+        }
+        if (NearProcessQueue.Dequeue(DequeuedChunk)
             || ProcessQueue.Dequeue(DequeuedChunk))
         {
             RecordStreamingResultDequeued(DequeuedChunk);
@@ -10104,6 +10134,10 @@ void AVoxelWorld::ProcessPendingChunks()
         if (bMeshApplied)
         {
             ++MeshesApplied;
+            if (bDequeuedSupplementalFar && ResultLevel >= 3)
+            {
+                ++SupplementalFarApplies;
+            }
             if (ResultLevel >= 0 && ResultLevel < 9)
             {
                 ++MeshesAppliedByLevel[ResultLevel];
@@ -10122,6 +10156,19 @@ void AVoxelWorld::ProcessPendingChunks()
         }
         if (MeshesApplied >= MaxApplies)
         {
+            const double DrainElapsedSeconds = FPlatformTime::ToSeconds64(
+                FPlatformTime::Cycles64() - DrainStartCycles);
+            const bool bCanSupplementFar = MeshesApplied == MaxApplies
+                && SupplementalFarApplies == 0
+                && !bUseStartupApplyBoost
+                && GVoxelForgeStreamingFarApplySupplement != 0
+                && ApplyBudgetSeconds > 0.0
+                && DrainElapsedSeconds < ApplyBudgetSeconds * 0.5;
+            if (bCanSupplementFar)
+            {
+                bForceSupplementalFarDequeue = true;
+                continue;
+            }
             break;
         }
     }
@@ -15360,6 +15407,8 @@ bool AVoxelWorld::ApplyMeshToTile(FChunkResult& Result)
     const int32 NumIndices = Result.NumTriangles * 3;
     const int32 FirstCapIndex = FMath::Clamp(
         NumIndices - Result.NumCeilingTriangles * 3, 0, NumIndices);
+    TFuture<ERealtimeMeshProxyUpdateStatus> ProxyPublicationFuture;
+    bool bHasProxyPublicationFuture = false;
     {
         TRACE_CPUPROFILER_EVENT_SCOPE(VoxelForge_RMC_CreateSections);
         if (bHasGroundTris)
@@ -15369,9 +15418,10 @@ bool AVoxelWorld::ApplyMeshToTile(FChunkResult& Result)
             const FRealtimeMeshStreamRange GroundRange =
                 URealtimeMeshBlueprintFunctionLibrary::MakeStreamRange(
                     0, Result.NumVertices, 0, FirstCapIndex);
-            RTMesh->CreateSection(
+            ProxyPublicationFuture = RTMesh->CreateSection(
                 FRealtimeMeshSectionKey::CreateForPolyGroup(GroupKey, 0),
                 GroundConfig, GroundRange, /*bShouldCreateCollision*/ bWantsCollision);
+            bHasProxyPublicationFuture = true;
         }
         if (bHasCeilingTris)
         {
@@ -15380,11 +15430,48 @@ bool AVoxelWorld::ApplyMeshToTile(FChunkResult& Result)
             const FRealtimeMeshStreamRange CapRange =
                 URealtimeMeshBlueprintFunctionLibrary::MakeStreamRange(
                     0, Result.NumVertices, FirstCapIndex, NumIndices);
-            RTMesh->CreateSection(
+            ProxyPublicationFuture = RTMesh->CreateSection(
                 FRealtimeMeshSectionKey::CreateForPolyGroup(GroupKey, 1),
                 CapConfig, CapRange, /*bShouldCreateCollision*/ bWantsCollision);
+            bHasProxyPublicationFuture = true;
         }
     }
+    const uint64 ProxyPublicationSubmitCycles = FPlatformTime::Cycles64();
+    if (bTraceRmcTiming && bHasProxyPublicationFuture)
+    {
+        const TWeakObjectPtr<AVoxelWorld> WeakWorld(this);
+        const TWeakObjectPtr<URealtimeMeshComponent> WeakComponent(MeshComp);
+        const FVoxelTileKey PublishedTile = Tile;
+        const uint64 RequestOrdinal = Result.RequestOrdinal;
+        const uint64 ApplyStartedCycles = ApplyStartCycles;
+        ProxyPublicationFuture.Next(
+            [WeakWorld, WeakComponent, PublishedTile, RequestOrdinal,
+             ApplyStartedCycles, ProxyPublicationSubmitCycles](
+                ERealtimeMeshProxyUpdateStatus Status)
+            {
+                if (!VoxelForgeStartupTrace::IsActive()) return;
+                const uint64 PublishedCycles = FPlatformTime::Cycles64();
+                const URealtimeMeshComponent* Component = WeakComponent.Get();
+                VoxelForgeStartupTrace::RecordEvent(TEXT("rmc_proxy_published"), FString::Printf(
+                    TEXT("\"tile\":[%d,%d,%d],\"level\":%d,\"request_ordinal\":%llu,"
+                         "\"proxy_update_status\":%d,\"apply_to_proxy_ms\":%.3f,"
+                         "\"section_submit_to_proxy_ms\":%.3f,\"component_visible\":%d,"
+                         "\"world_valid\":%d"),
+                    PublishedTile.Coord.X, PublishedTile.Coord.Y, PublishedTile.Coord.Z,
+                    PublishedTile.Level, static_cast<unsigned long long>(RequestOrdinal),
+                    static_cast<int32>(Status),
+                    ApplyStartedCycles != 0 && PublishedCycles >= ApplyStartedCycles
+                        ? FPlatformTime::ToSeconds64(PublishedCycles - ApplyStartedCycles) * 1000.0
+                        : 0.0,
+                    ProxyPublicationSubmitCycles != 0
+                            && PublishedCycles >= ProxyPublicationSubmitCycles
+                        ? FPlatformTime::ToSeconds64(
+                              PublishedCycles - ProxyPublicationSubmitCycles) * 1000.0
+                        : 0.0,
+                    Component && Component->IsVisible() ? 1 : 0,
+                    WeakWorld.IsValid() ? 1 : 0));
+            });
+    }
     if (bTraceRmcTiming)
     {
         TraceSectionConfigMilliseconds = FPlatformTime::ToSeconds64(

Source/VoxelForge/Public/VoxelStartupTrace.h

@@ -59,6 +59,7 @@ namespace VoxelForgeStartupTrace
         int32 ResultsDrained = 0;
         int32 ApplyResultsByLevelTotal = 0;
         int32 MeshesApplied = 0;
+        int32 SupplementalFarApplies = 0;
         int32 MaxApplies = 0;
         double ApplyBudgetMilliseconds = 0.0;
         bool bCountLimitReached = false;

Standalone/Research/WorldAssemblerM3/Linux121/REPORT.md

@@ -0,0 +1,26 @@
+# Linux 121 — finished far tiles
+
+One change: after four normal visible applies, allow one far apply if less than half of the 2 ms budget is used. Disable the lane during startup boost. Geometry and worker limits stay unchanged.
+
+Five interleaved UE pairs at 30 m/s (8 slots, ready depth 8) reduced median far coverage holes from 4.38% to 2.70%; near settled coverage rose from 45.12% to 45.43%. Far worker-finish-to-apply wait fell from 1.105 s / 3.25 s p50/p95 to 0.114 s / 2.40 s. The lane applied 672 extra far tiles across the five runs.
+
+The worst tick was 14.74 ms, below L120's 15.43 ms ceiling. The worst per-run p99 was 11.00 ms versus 11.58 ms for baseline. Five 140 cm/s arrival pairs improved paired median LOD0–2 settlement by 20 ms and whole-view/far-clear by 313 ms. Both arms had zero far coverage holes at 20 s.
+
+## Stage timings
+
+Five paired UE traces per arm. Result wait is worker finish to `ApplyTileResult` (p50/p95 ms); generation, apply, and proxy are p95 ms. Proxy lag is reconstructed from callback and tile-apply timestamps because profile-only proxy-lag fields are zero when profiling is off.
+
+| Speed / LOD | Far generation p95 | Result wait baseline → lane | Apply baseline → lane | Proxy baseline → lane |
+|---|---:|---:|---:|---:|
+| 1.4 m/s near | — | 12 / 43 → 9 / 37 | 0.362 → 0.340 | 0.411 → 0.406 |
+| 1.4 m/s far | 5.6 → 6.3 | 209 / 229 → 142 / 209 | 0.314 → 0.252 | 0.221 → 0.196 |
+| 30 m/s near | — | 12 / 125 → 12 / 120 | 0.206 → 0.197 | 0.250 → 0.237 |
+| 30 m/s far | 10.0 → 10.3 | 1,105 / 3,250 → 114 / 2,397 | 0.351 → 0.324 | 0.335 → 0.312 |
+
+The four-apply count limit was hit on 44% of baseline moving frames; the 2 ms time limit was hit on under 1%. Stale and obsolete results were discarded before apply and were under 1% of tile events.
+
+Admission had idle-worker/queued-result windows while far coverage was missing: 504/2,135 baseline and 446/2,109 lane samples. The 256-result cap did not cause those windows: backlog p95/max was 101/206, with zero cap declines in them. Core-region compilation was pending in 64%/62% of those windows.
+
+RMC reported `Updated` after proxy publication, but many replacement components were still hidden by the coverage handoff. NullRHI runs cannot measure the first presented pixel; the trace's far coverage is explicitly not visual enclosure.
+
+UE build passed with `-NoUBA -MaxParallelActions=4`; fast tier passed (269 gates, 5 diagnostics). Clang/GCC W1/W8/W12 matched L104 across four strata: 24/24 exact tile and whole-view hashes. Headless has no UE apply budget, so the lane has no headless effect. Logs are under `Saved/Linux/l121`; timing used CPUs 0–11 with no build running.

Standalone/Research/WorldAssemblerM3/Linux121/RunL121Arrival.py

@@ -0,0 +1,488 @@
+#!/usr/bin/env python3
+"""Run one pinned null-RHI FI arrival for Linux 121."""
+from __future__ import annotations
+
+import argparse
+from collections import Counter
+from datetime import datetime
+import json
+import re
+import subprocess
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+HOST = Path("/home/jahni/VFHost")
+HOST_PLUGIN_SAVED = HOST / "Plugins/VoxelForge/Saved/Linux/l121"
+HOST_SAVED = HOST / "Saved/Linux/l121"
+EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd")
+OUT = ROOT / "Saved/Linux/l121/arrival"
+SPAWNS_M = {
+    "FI": (25.8369140625, 3.19921875, -737.0),
+    "SW": (25.8369140625, 3.19921875, -200.0),
+}
+MOVE_START_S = 20.0
+MOVE_DURATION_S = 12.0
+EXIT_S = MOVE_START_S + MOVE_DURATION_S + 5.0
+RUN_TIMEOUT_S = 1740
+
+
+def read_events(path: Path) -> list[dict]:
+    events = []
+    if path.is_file():
+        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+            if line.startswith("{"):
+                try:
+                    events.append(json.loads(line))
+                except json.JSONDecodeError:
+                    pass
+    return events
+
+
+def percentile(values: list[float], fraction: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    return ordered[min(len(ordered) - 1, int(round((len(ordered) - 1) * fraction)))]
+
+
+def sample_unreal_peak_rss() -> tuple[int, int] | None:
+    executable = str(EDITOR).encode()
+    project = str(HOST / "VFHost.uproject").encode()
+    peak: tuple[int, int] | None = None
+    for process in Path("/proc").glob("[0-9]*"):
+        try:
+            command_line = (process / "cmdline").read_bytes()
+            if executable not in command_line or project not in command_line:
+                continue
+            status = (process / "status").read_text(encoding="ascii", errors="ignore")
+            match = re.search(r"^VmHWM:\s+(\d+)\s+kB$", status, re.MULTILINE)
+            if match:
+                rss_kib = int(match.group(1))
+                if peak is None or rss_kib > peak[0]:
+                    peak = (rss_kib, int(process.name))
+        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
+            continue
+    return peak
+
+
+def endphysics_gap(stdout: Path) -> tuple[float | None, int | None]:
+    release_times: dict[int, dict[int, datetime]] = {}
+    pattern = re.compile(
+        r"^\[(\d{4}\.\d{2}\.\d{2}-\d{2}\.\d{2}\.\d{2}:\d{3})\].*"
+        r"LogTick: tick\s+(\d+)\s+-+\s+Release tick group ([34])")
+    with stdout.open(encoding="utf-8", errors="replace") as stream:
+        for line in stream:
+            match = pattern.match(line)
+            if not match:
+                continue
+            timestamp = datetime.strptime(match.group(1), "%Y.%m.%d-%H.%M.%S:%f")
+            frame = int(match.group(2))
+            group = int(match.group(3))
+            release_times.setdefault(frame, {})[group] = timestamp
+    gaps = [(round((times[4] - times[3]).total_seconds() * 1000.0, 3), frame)
+            for frame, times in release_times.items() if 3 in times and 4 in times]
+    return max(gaps, default=(None, None))
+
+
+def run(stratum: str, repeat: int, phase: str, ready_depth: int,
+        background_cap: int, worker_reserve: int, helper_background: int,
+        tile_background_low: int, tile_background_normal: int, startup_boost: int,
+        tick_logs: bool, insights: bool, far_apply_supplement: int) -> dict:
+    label = (f"{phase}-{stratum.lower()}-rd{ready_depth}-bg{background_cap}"
+             f"-reserve{worker_reserve}-helperbg{helper_background}"
+             f"-tilelow{tile_background_low}-tilenormal{tile_background_normal}"
+             f"-ticklog{int(tick_logs)}"
+             f"-farapply{far_apply_supplement}-boost{startup_boost}-{repeat}")
+    trace = OUT / f"{label}.trace.jsonl"
+    utrace = OUT / f"{label}.utrace"
+    abslog = OUT / f"{label}.log"
+    stdout = OUT / f"{label}.stdout.log"
+    output = HOST_PLUGIN_SAVED / "output" / label
+    userdir = HOST_SAVED / "user" / label
+    zendata = HOST_SAVED / "zen" / label
+    ddc = HOST_SAVED / "DDC"
+    for directory in (OUT, output, userdir, zendata, ddc):
+        directory.mkdir(parents=True, exist_ok=True)
+    trace.unlink(missing_ok=True)
+    utrace.unlink(missing_ok=True)
+    spawn_cm = ",".join(f"{value * 100.0:.3f}" for value in SPAWNS_M[stratum])
+    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", f"-slot={1 if stratum == 'FI' else 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={output}", "-voxel.StreamingCoarseFirst=1",
+        "-voxel.StreamingDynamicNearReserve=1",
+        "-voxel.StreamingLegacyPendingResultCap=0",
+        "-voxel.CoreRegionConcurrentAdmission=1",
+        "-voxel.StreamingResultBacklogCap=256",
+        f"-voxel.StreamingReadyQueueDepth={ready_depth}",
+        f"-voxel.StreamingBackgroundWorkerCap={background_cap}",
+        f"-voxel.StreamingBackgroundWorkerReserve={worker_reserve}",
+        f"-voxel.CoreParallelForBackgroundPriority={helper_background}",
+        f"-voxel.StreamingTileBackgroundLow={tile_background_low}",
+        f"-voxel.StreamingTileBackgroundNormal={tile_background_normal}",
+        f"-voxel.TestLogTickFunctions={int(tick_logs)}",
+        "-voxel.CoreColumnCacheDirectMapped=0", "-voxel.CoreMCGradientMemoization=1",
+        "-voxel.StreamingCollisionLookahead=1",
+        f"-voxel.StreamingStartupApplyBoost={startup_boost}",
+        f"-voxel.StreamingFarApplySupplement={far_apply_supplement}",
+        "-voxel.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=140.0",
+        f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}",
+        f"-voxel.TestMoveDurationSeconds={MOVE_DURATION_S:.1f}",
+        f"-voxel.TestExitSeconds={EXIT_S:.1f}", "-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 insights:
+        command += ["-trace=cpu,task,frame", "-statnamedevents",
+                    f"-tracefile={utrace}", "-tracefiletrunc"]
+    wrapped = ["taskset", "-c", "0-11", "timeout", f"{RUN_TIMEOUT_S}s", *command]
+    started = time.monotonic()
+    with stdout.open("w", encoding="utf-8") as stream:
+        process = subprocess.Popen(wrapped, cwd=ROOT, stdout=stream,
+                                   stderr=subprocess.STDOUT)
+        peak_rss = None
+        while process.poll() is None:
+            observed = sample_unreal_peak_rss()
+            if observed and (peak_rss is None or observed[0] > peak_rss[0]):
+                peak_rss = observed
+            time.sleep(0.25)
+        exit_code = process.wait()
+        observed = sample_unreal_peak_rss()
+        if observed and (peak_rss is None or observed[0] > peak_rss[0]):
+            peak_rss = observed
+
+    events = read_events(trace)
+    worker_backend_config = next(
+        (e for e in events if e.get("type") == "worker_backend_config"), {})
+    worker_cap_readback = next(
+        (e for e in events if e.get("type") == "streaming_worker_cap_readback"), {})
+    samples = [e for e in events if e.get("type") == "streaming_sample"]
+    arrival = [e for e in samples if float(e.get("t_s", 0.0)) < MOVE_START_S]
+    full = next((e for e in arrival
+                 if sum(e.get("desired_by_lod", [0] * 9)[:3]) > 0
+                 and sum(e.get("settled_by_lod", [0] * 9)[:3])
+                 >= sum(e.get("desired_by_lod", [0] * 9)[:3])), None)
+    first_admission = next((e for e in events if e.get("type") == "scheduler_admission"
+                            and e.get("desired_rebuilt") == 1), None)
+    request_events = [e for e in events if e.get("type") == "request"]
+    initial_request_events = [e for e in request_events if e.get("epoch") == 0]
+    tile_by_ordinal = {e.get("request_ordinal"): e for e in events
+                       if e.get("type") == "tile"}
+    collision_count = sum(1 for e in events if e.get("type") == "collision_ready"
+                          and e.get("level", 9) <= 2
+                          and full and e.get("t_s", 1e30) <= full["t_s"])
+    initial_tiles = [tile_by_ordinal[r.get("request_ordinal")] for r in initial_request_events
+                     if full and r.get("t_s", 1e30) <= full["t_s"]
+                     and r.get("request_ordinal") in tile_by_ordinal
+                     and tile_by_ordinal[r["request_ordinal"]].get("level", 9) <= 2]
+    burst_start_s = float(first_admission["t_s"]) if first_admission else -1.0
+    burst_end_s = float(full["t_s"]) if full else -1.0
+    work_by_frame = {int(e.get("frame", -1)): e for e in events
+                     if e.get("type") == "game_thread_work_frame"}
+    work_values = [e for e in work_by_frame.values()
+                   if burst_start_s <= float(e.get("t_s", -1.0)) <= burst_end_s]
+    arrival_frame_ids = {int(e.get("frame", -1)) for e in work_values}
+    scheduler_by_frame = {int(e.get("frame", -1)): e for e in events
+                          if e.get("type") == "scheduler_admission"}
+    scheduler_values = [e for e in events if e.get("type") == "scheduler_admission"
+                        and burst_start_s <= float(e.get("t_s", -1.0)) <= burst_end_s]
+    apply_by_frame = {int(e.get("frame", -1)): e for e in events
+                      if e.get("type") == "apply_frame"}
+    commandlet_tick_by_frame = {int(e.get("frame", -1)): e for e in events
+                                if e.get("type") == "commandlet_world_tick_frame"}
+    garbage_collect_by_frame = {int(e.get("frame", -1)): e for e in events
+                                if e.get("type") == "garbage_collect"}
+    # apply_frame N+1 records the interval containing AVoxelWorld::Tick N. Its apply_ms
+    # belongs to Tick N+1, so use apply_by_frame[N] for the same-tick apply breakdown.
+    near_intervals = [apply_by_frame[frame + 1] for frame in arrival_frame_ids
+                      if frame + 1 in apply_by_frame]
+    rmc_by_frame: dict[int, dict[str, float]] = {}
+    for event in events:
+        frame = int(event.get("frame", -1))
+        if event.get("type") != "rmc_submit" or frame not in arrival_frame_ids:
+            continue
+        values = rmc_by_frame.setdefault(frame, {
+            "component_acquire_ms": 0.0,
+            "create_section_group_ms": 0.0,
+            "create_sections_ms": 0.0,
+            "collision_handoff_ms": 0.0,
+        })
+        for key in values:
+            values[key] += float(event.get(key, 0.0))
+    collision_completion_by_frame: dict[int, float] = {}
+    for event in events:
+        frame = int(event.get("frame", -1))
+        if event.get("type") == "collision_completion" and frame in arrival_frame_ids:
+            collision_completion_by_frame[frame] = collision_completion_by_frame.get(frame, 0.0) \
+                + float(event.get("handler_ms", 0.0))
+    near_requests = [r for r in request_events if full and r.get("t_s", 1e30) <= full["t_s"]
+                     and r.get("level", 9) <= 2]
+    initial_lod0_counts = Counter(
+        tuple(r.get("tile", ())) for r in near_requests if r.get("level") == 0)
+    repeated_lod0_keys = [
+        {"tile": list(tile), "requests": count}
+        for tile, count in sorted(initial_lod0_counts.items()) if count > 1]
+    initial_scheduler_frames = [e for e in events
+        if e.get("type") == "scheduler_admission" and full
+        and e.get("t_s", 1e30) <= full["t_s"]]
+    worst_interval = max(near_intervals, key=lambda e: float(e.get("frame_ms", 0.0)), default=None)
+    worst_interval_frame_id = int(worst_interval.get("frame", -1)) if worst_interval else -1
+    worst_work_frame_id = worst_interval_frame_id - 1
+    worst_work = work_by_frame.get(worst_work_frame_id, {})
+    worst_apply = apply_by_frame.get(worst_work_frame_id, {})
+    worst_scheduler = scheduler_by_frame.get(worst_work_frame_id, {})
+    worst_rmc = rmc_by_frame.get(worst_work_frame_id, {})
+    worst_collision_completion_ms = collision_completion_by_frame.get(worst_work_frame_id, 0.0)
+    rmc_values = [rmc_by_frame.get(frame, {}) for frame in arrival_frame_ids]
+    apply_values = [apply_by_frame[frame] for frame in arrival_frame_ids
+                    if frame in apply_by_frame]
+    commandlet_tick_values = [commandlet_tick_by_frame[frame] for frame in arrival_frame_ids
+                              if frame in commandlet_tick_by_frame]
+    garbage_collect_values = [garbage_collect_by_frame[frame] for frame in arrival_frame_ids
+                              if frame in garbage_collect_by_frame]
+
+    def work_stat(name: str, fraction: float | None = None) -> float | None:
+        values = [float(e.get(name, 0.0)) for e in work_values]
+        return percentile(values, fraction) if fraction is not None else max(values, default=0.0)
+
+    def scheduler_stat(name: str, fraction: float | None = None) -> float | None:
+        values = [float(e.get(name, 0.0)) for e in scheduler_values]
+        return percentile(values, fraction) if fraction is not None else max(values, default=0.0)
+
+    def rmc_stat(name: str, fraction: float | None = None) -> float | None:
+        values = [float(e.get(name, 0.0)) for e in rmc_values]
+        return percentile(values, fraction) if fraction is not None else max(values, default=0.0)
+
+    def commandlet_stat(name: str, fraction: float | None = None) -> float | None:
+        values = [float(e.get(name, 0.0)) for e in commandlet_tick_values]
+        return percentile(values, fraction) if fraction is not None else max(values, default=0.0)
+
+    row = {
+        "label": label, "stratum": stratum, "repeat": repeat, "phase": phase,
+        "ready_queue_depth": ready_depth,
+        "requested_background_worker_cap": background_cap,
+        "background_worker_cap": worker_cap_readback.get(
+            "streaming_background_worker_cap", background_cap),
+        "requested_background_worker_reserve": worker_reserve,
+        "background_worker_reserve": worker_backend_config.get(
+            "streaming_background_worker_reserve", worker_cap_readback.get(
+                "background_worker_reserve", worker_reserve)),
+        "requested_core_parallel_for_background_priority": helper_background,
+        "core_parallel_for_background_priority": worker_backend_config.get(
+            "core_parallel_for_background_priority", helper_background),
+        "requested_streaming_tile_background_low": tile_background_low,
+        "streaming_tile_background_low": worker_backend_config.get(
+            "streaming_tile_background_low", tile_background_low),
+        "requested_streaming_tile_background_normal": tile_background_normal,
+        "streaming_tile_background_normal": worker_backend_config.get(
+            "streaming_tile_background_normal", tile_background_normal),
+        "effective_tile_slots": worker_backend_config.get("effective_tile_slots"),
+        "background_worker_threads": worker_backend_config.get("taskgraph_background_threads"),
+        "startup_apply_boost": startup_boost,
+        "streaming_far_apply_supplement": worker_backend_config.get(
+            "streaming_far_apply_supplement"),
+        "insights": insights,
+        "engine_tick_logs": tick_logs,
+        "exit_code": exit_code,
+        "launch_wrapper": "taskset -c 0-11 timeout 1740s (no address-space limit)",
+        "peak_rss_kib": peak_rss[0] if peak_rss else None,
+        "peak_rss_mib": round(peak_rss[0] / 1024.0, 1) if peak_rss else None,
+        "peak_rss_pid": peak_rss[1] if peak_rss else None,
+        "endphysics_to_postphysics_gap_ms_max": None,
+        "endphysics_gap_frame": None,
+        "wall_s": round(time.monotonic() - started, 3),
+        "desired_t_s": next((e.get("t_s") for e in events if e.get("type") == "desired_set"), None),
+        "first_admission_t_s": first_admission.get("t_s") if first_admission else None,
+        "lod02_full_settled_t_s": full.get("t_s") if full else None,
+        "near_settle_from_first_admission_s": round(
+            float(full["t_s"]) - float(first_admission["t_s"]), 4)
+            if full and first_admission else None,
+        "desired_by_lod": full.get("desired_by_lod", [])[:3] if full else None,
+        "settled_by_lod": full.get("settled_by_lod", [])[:3] if full else None,
+        "near_admitted_tiles": len(near_requests),
+        "initial_lod0_admissions": sum(initial_lod0_counts.values()),
+        "repeated_lod0_request_keys_before_settle": repeated_lod0_keys,
+        "peak_result_backlog_before_settle": max(
+            (int(e.get("result_backlog", 0)) for e in initial_scheduler_frames), default=0),
+        "last_admission_t_s": max((e.get("t_s", 0.0) for e in near_requests), default=None),
+        "near_tile_events": len(initial_tiles),
+        "near_worker_seconds": round(sum(e.get("worker_task_s", 0.0) for e in initial_tiles), 4),
+        "near_core_seconds": round(sum(e.get("core_generate_s", 0.0) for e in initial_tiles), 4),
+        "collision_ready_by_settle": collision_count,
+        "burst_frame_count": len(near_intervals),
+        "burst_frame_ms_p95": round(sorted(e.get("frame_ms", 0.0) for e in near_intervals)[
+            min(len(near_intervals) - 1, int(.95 * (len(near_intervals) - 1)))], 3)
+            if near_intervals else None,
+        "burst_frame_ms_max": round(max((e.get("frame_ms", 0.0) for e in near_intervals), default=0.0), 3),
+        "burst_full_world_tick_ms_p99": round(commandlet_stat("world_tick_ms", .99) or 0.0, 3),
+        "burst_full_world_tick_ms_max": round(commandlet_stat("world_tick_ms") or 0.0, 3),
+        "burst_world_pre_actor_ms_p99": round(commandlet_stat("world_pre_actor_ms", .99) or 0.0, 3),
+        "burst_world_pre_actor_ms_max": round(commandlet_stat("world_pre_actor_ms") or 0.0, 3),
+        "burst_world_actor_phase_ms_p99": round(commandlet_stat("world_actor_phase_ms", .99) or 0.0, 3),
+        "burst_world_actor_phase_ms_max": round(commandlet_stat("world_actor_phase_ms") or 0.0, 3),
+        "burst_world_post_actor_ms_p99": round(commandlet_stat("world_post_actor_ms", .99) or 0.0, 3),
+        "burst_world_post_actor_ms_max": round(commandlet_stat("world_post_actor_ms") or 0.0, 3),
+        "burst_commandlet_readiness_check_ms_max": round(
+            commandlet_stat("readiness_check_ms") or 0.0, 3),
+        "burst_commandlet_sleep_ms_p99": round(commandlet_stat("sleep_ms", .99) or 0.0, 3),
+        "burst_commandlet_sleep_ms_max": round(commandlet_stat("sleep_ms") or 0.0, 3),
+        "burst_commandlet_loop_ms_max": round(commandlet_stat("loop_ms") or 0.0, 3),
+        "burst_gc_count": len(garbage_collect_values),
+        "burst_gc_duration_ms_max": round(max(
+            (float(e.get("duration_ms", 0.0)) for e in garbage_collect_values), default=0.0), 3),
+        "burst_max_meshes_per_frame": max((e.get("meshes_applied", 0) for e in apply_values), default=0),
+        "burst_max_apply_ms": round(max((e.get("apply_ms", 0.0) for e in apply_values), default=0.0), 3),
+        "burst_max_admit_per_frame": max((sum(e.get("submitted_by_lod", [0] * 9)[:3])
+                                            for e in scheduler_values), default=0),
+        "burst_frame_timing_count": len(work_values),
+        "burst_active_generation_workers_p95": round(work_stat("active_generation_workers", .95) or 0.0, 3),
+        "burst_active_generation_workers_max": round(work_stat("active_generation_workers") or 0.0, 3),
+        "burst_active_tile_tasks_max": round(work_stat("active_tile_tasks") or 0.0, 3),
+        "burst_world_tick_ms_p99": round(work_stat("tick_ms", .99) or 0.0, 3),
+        "burst_world_tick_ms_max": round(work_stat("tick_ms") or 0.0, 3),
+        "burst_update_chunks_ms_p99": round(work_stat("update_chunks_ms", .99) or 0.0, 3),
+        "burst_update_chunks_ms_max": round(work_stat("update_chunks_ms") or 0.0, 3),
+        "burst_pending_drain_ms_p99": round(work_stat("pending_drain_ms", .99) or 0.0, 3),
+        "burst_pending_drain_ms_max": round(work_stat("pending_drain_ms") or 0.0, 3),
+        "burst_post_apply_callbacks_ms_p99": round(work_stat("post_apply_callbacks_ms", .99) or 0.0, 3),
+        "burst_post_apply_callbacks_ms_max": round(work_stat("post_apply_callbacks_ms") or 0.0, 3),
+        "burst_visibility_reconcile_ms_p99": round(work_stat("visibility_reconcile_ms", .99) or 0.0, 3),
+        "burst_visibility_reconcile_ms_max": round(work_stat("visibility_reconcile_ms") or 0.0, 3),
+        "burst_collision_probe_ms_max": round(work_stat("collision_probe_ms") or 0.0, 3),
+        "burst_visible_apply_probe_ms_max": round(work_stat("visible_apply_probe_ms") or 0.0, 3),
+        "burst_lod0_ready_ms_max": round(work_stat("lod0_ready_ms") or 0.0, 3),
+        "burst_steady_trace_ms_max": round(work_stat("steady_trace_ms") or 0.0, 3),
+        "burst_modification_ms_max": round(work_stat("modification_ms") or 0.0, 3),
+        "burst_lifecycle_audit_ms_p99": round(work_stat("lifecycle_audit_ms", .99) or 0.0, 3),
+        "burst_lifecycle_audit_ms_max": round(work_stat("lifecycle_audit_ms") or 0.0, 3),
+        "burst_collision_gate_ms_p99": round(work_stat("collision_gate_ms", .99) or 0.0, 3),
+        "burst_collision_gate_ms_max": round(work_stat("collision_gate_ms") or 0.0, 3),
+        "burst_unload_ms_p99": round(work_stat("unload_ms", .99) or 0.0, 3),
+        "burst_unload_ms_max": round(work_stat("unload_ms") or 0.0, 3),
+        "burst_trace_json_ms_p99": round(work_stat("trace_json_ms", .99) or 0.0, 3),
+        "burst_trace_json_ms_max": round(work_stat("trace_json_ms") or 0.0, 3),
+        "burst_desired_rebuild_ms_max": round(scheduler_stat("desired_rebuild_ms") or 0.0, 3),
+        "burst_rmc_create_group_ms_p99": round(rmc_stat("create_section_group_ms", .99) or 0.0, 3),
+        "burst_rmc_create_group_ms_max": round(rmc_stat("create_section_group_ms") or 0.0, 3),
+        "burst_rmc_collision_handoff_ms_max": round(rmc_stat("collision_handoff_ms") or 0.0, 3),
+        "burst_collision_completion_ms_max": round(max(
+            collision_completion_by_frame.values(), default=0.0), 3),
+        "worst_frame_breakdown": {
+            "frame": worst_work_frame_id,
+            "world_tick_frame": worst_work_frame_id,
+            "frame_interval_end_frame": worst_interval_frame_id,
+            "frame_interval_ms": round(float(worst_interval.get("frame_ms", 0.0)), 3)
+                if worst_interval else None,
+            "apply_ms": round(float(worst_apply.get("apply_ms", 0.0)), 3)
+                if worst_apply else None,
+            "result_drain_ms": round(float(worst_apply.get("drain_ms", 0.0)), 3)
+                if worst_apply else None,
+            "outside_world_tick_ms": round(max(0.0, float(worst_interval.get("frame_ms", 0.0))
+                - float(worst_work.get("tick_ms", 0.0))), 3)
+                if worst_interval else None,
+            "outside_full_world_tick_ms": round(max(0.0, float(worst_interval.get("frame_ms", 0.0))
+                - float(commandlet_tick_by_frame.get(worst_work_frame_id, {}).get("world_tick_ms", 0.0))), 3)
+                if worst_interval else None,
+            "full_world_tick_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("world_tick_ms", 0.0)), 3),
+            "world_pre_actor_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("world_pre_actor_ms", 0.0)), 3),
+            "world_actor_phase_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("world_actor_phase_ms", 0.0)), 3),
+            "world_post_actor_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("world_post_actor_ms", 0.0)), 3),
+            "commandlet_readiness_check_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("readiness_check_ms", 0.0)), 3),
+            "commandlet_sleep_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("sleep_ms", 0.0)), 3),
+            "commandlet_loop_ms": round(float(commandlet_tick_by_frame.get(
+                worst_work_frame_id, {}).get("loop_ms", 0.0)), 3),
+            "gc_duration_ms": round(float(garbage_collect_by_frame.get(
+                worst_work_frame_id, {}).get("duration_ms", 0.0)), 3),
+            "world_tick_ms": round(float(worst_work.get("tick_ms", 0.0)), 3),
+            "active_generation_workers": int(worst_work.get("active_generation_workers", 0)),
+            "active_tile_tasks": int(worst_work.get("active_tile_tasks", 0)),
+            "collision_gate_ms": round(float(worst_work.get("collision_gate_ms", 0.0)), 3),
+            "update_chunks_ms": round(float(worst_work.get("update_chunks_ms", 0.0)), 3),
+            "desired_rebuild_ms": round(float(worst_scheduler.get("desired_rebuild_ms", 0.0)), 3),
+            "pending_drain_ms": round(float(worst_work.get("pending_drain_ms", 0.0)), 3),
+            "post_apply_callbacks_ms": round(float(worst_work.get("post_apply_callbacks_ms", 0.0)), 3),
+            "collision_probe_ms": round(float(worst_work.get("collision_probe_ms", 0.0)), 3),
+            "visibility_reconcile_ms": round(float(worst_work.get("visibility_reconcile_ms", 0.0)), 3),
+            "visible_apply_probe_ms": round(float(worst_work.get("visible_apply_probe_ms", 0.0)), 3),
+            "lod0_ready_ms": round(float(worst_work.get("lod0_ready_ms", 0.0)), 3),
+            "unload_ms": round(float(worst_work.get("unload_ms", 0.0)), 3),
+            "steady_trace_ms": round(float(worst_work.get("steady_trace_ms", 0.0)), 3),
+            "modification_ms": round(float(worst_work.get("modification_ms", 0.0)), 3),
+            "lifecycle_audit_ms": round(float(worst_work.get("lifecycle_audit_ms", 0.0)), 3),
+            "trace_json_ms": round(float(worst_work.get("trace_json_ms", 0.0)), 3),
+            "trace_archive_serialize_ms": round(float(
+                worst_work.get("trace_archive_serialize_ms", 0.0)), 3),
+            "collision_completion_ms": round(worst_collision_completion_ms, 3),
+            **{key: round(float(worst_rmc.get(key, 0.0)), 3)
+               for key in ("component_acquire_ms", "create_section_group_ms",
+                           "create_sections_ms", "collision_handoff_ms")},
+        },
+        "trace": str(trace.relative_to(ROOT)), "utrace": str(utrace.relative_to(ROOT)),
+        "abslog": str(abslog.relative_to(ROOT)), "stdout": str(stdout.relative_to(ROOT)),
+    }
+    stdout_text = stdout.read_text(encoding="utf-8", errors="replace")
+    gap_ms, gap_frame = endphysics_gap(stdout)
+    row["endphysics_to_postphysics_gap_ms_max"] = gap_ms
+    row["endphysics_gap_frame"] = gap_frame
+    failure = re.search(r"\[VoxelForgeCoreWorldTest\] failed: ([^\r\n]+)", stdout_text)
+    if failure:
+        row["world_test_failure"] = failure.group(1)
+    with (OUT / "runs.jsonl").open("a", encoding="utf-8") as stream:
+        stream.write(json.dumps(row, sort_keys=True) + "\n")
+    (OUT / f"{label}.summary.json").write_text(json.dumps(row, indent=2) + "\n",
+                                                 encoding="utf-8")
+    print("[L105UnrealArrival] " + json.dumps(row, sort_keys=True), flush=True)
+    if exit_code:
+        raise SystemExit(f"Unreal commandlet failed: {stdout}; exit={exit_code}")
+    if not full or len(arrival) < 8:
+        raise SystemExit(f"Unreal arrival trace incomplete: {stdout}; exit={exit_code}")
+    if insights and (not utrace.is_file() or utrace.stat().st_size < 1024):
+        raise SystemExit(f"Unreal Insights trace missing or empty: {utrace}")
+    return row
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--stratum", choices=("FI", "SW"), required=True)
+    parser.add_argument("--repeat", type=int, default=1)
+    parser.add_argument("--phase", choices=("probe", "before", "after", "snapshot"), default="probe")
+    parser.add_argument("--ready-depth", type=int, default=64)
+    parser.add_argument("--background-cap", type=int, choices=(0, 1), default=1)
+    parser.add_argument("--worker-reserve", type=int, choices=(0, 1, 2, 3), default=0)
+    parser.add_argument("--helper-background", type=int, choices=(0, 1), default=1)
+    parser.add_argument("--tile-background-low", type=int, choices=(0, 1), default=1)
+    parser.add_argument("--tile-background-normal", type=int, choices=(0, 1), default=0)
+    parser.add_argument("--startup-boost", type=int, choices=(0, 1), default=1)
+    parser.add_argument("--tick-logs", action="store_true")
+    parser.add_argument("--insights", action="store_true")
+    args = parser.parse_args()
+    if not EDITOR.is_file():
+        raise SystemExit(f"missing Unreal commandlet executable: {EDITOR}")
+    run(args.stratum, args.repeat, args.phase, args.ready_depth,
+        args.background_cap, args.worker_reserve, args.helper_background,
+        args.tile_background_low, args.tile_background_normal, args.startup_boost,
+        args.tick_logs, args.insights)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

Standalone/Research/WorldAssemblerM3/Linux121/RunL121ArrivalPairs.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Five interleaved FI arrival pairs, comparing normal applies with one supplemental far apply."""
+from __future__ import annotations
+
+import importlib.util
+import argparse
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+RUNNER_PATH = Path(__file__).with_name("RunL121Arrival.py")
+SPEC = importlib.util.spec_from_file_location("l121_arrival", RUNNER_PATH)
+assert SPEC and SPEC.loader
+ARRIVAL = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(ARRIVAL)
+
+
+def add_view_metrics(row: dict) -> dict:
+    trace_path = ROOT / row["trace"]
+    events = ARRIVAL.read_events(trace_path)
+    config = next((e for e in events if e.get("type") == "worker_backend_config"), {})
+    row["effective_tile_slots"] = config.get("effective_tile_slots")
+    row["near_priority"] = config.get("near_priority")
+    row["horizon_priority"] = config.get("horizon_priority")
+    samples = [e for e in events if e.get("type") == "streaming_sample"
+               and float(e.get("t_s", 0.0)) < ARRIVAL.MOVE_START_S
+               and float(e.get("speed_cm_s", 0.0)) < 50.0]
+
+    def total(values: list[int]) -> int:
+        return sum(values)
+
+    def near_complete(event: dict) -> bool:
+        desired = event.get("desired_by_lod", [0] * 9)
+        settled = event.get("settled_by_lod", [0] * 9)
+        return total(desired[:3]) > 0 and total(settled[:3]) >= total(desired[:3])
+
+    def whole_complete(event: dict) -> bool:
+        desired = event.get("desired_by_lod", [0] * 9)
+        settled = event.get("settled_by_lod", [0] * 9)
+        return total(desired) > 0 and total(settled) >= total(desired)
+
+    def far_holes(event: dict) -> float | None:
+        desired = event.get("desired_by_lod", [0] * 9)[3:]
+        covered = event.get("covered_by_lod", [0] * 9)[3:]
+        denominator = total(desired)
+        return (100.0 * (denominator - total(covered)) / denominator
+                if denominator else None)
+
+    first_near = next((e for e in samples if near_complete(e)), None)
+    first_whole = next((e for e in samples if whole_complete(e)), None)
+    first_far_clear = next((e for e in samples if total(e.get("desired_by_lod", [0] * 9)[3:])
+                             and far_holes(e) == 0.0), None)
+    last_sample = samples[-1] if samples else None
+    row["lod02_settled_t_s"] = float(first_near["t_s"]) if first_near else None
+    row["whole_view_settled_t_s"] = float(first_whole["t_s"]) if first_whole else None
+    row["far_clear_t_s"] = float(first_far_clear["t_s"]) if first_far_clear else None
+    row["far_holes_at_20s_pct"] = far_holes(last_sample) if last_sample else None
+    row["arrival_sample_count"] = len(samples)
+    return row
+
+
+def run_condition(condition: str, repetition: int, ready_depth: int) -> dict:
+    cap = 1
+    expected_slots = 8
+    row = ARRIVAL.run(
+        "FI", repetition, condition, ready_depth, cap, 0, 1, 1, 0, 1, False, False,
+        0 if condition == "baseline" else 1)
+    row = add_view_metrics(row)
+    if row["effective_tile_slots"] != expected_slots:
+        raise RuntimeError(f"worker slot readback mismatch: {row}")
+    if row["streaming_far_apply_supplement"] != (0 if condition == "baseline" else 1):
+        raise RuntimeError(f"far apply supplement CVar mismatch: {row}")
+    return row
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--ready-depth", type=int, choices=(8, 64), default=8)
+    ready_depth = parser.parse_args().ready_depth
+    OUT = ROOT / "Saved/Linux/l121/arrival"
+    OUT.mkdir(parents=True, exist_ok=True)
+    results = []
+    for condition in ("baseline", "far_supplement"):
+        warmup = run_condition(condition, 0, ready_depth)
+        warmup["warmup"] = True
+        results.append(warmup)
+
+    for repetition in range(1, 6):
+        order = ("baseline", "far_supplement") if repetition % 2 else ("far_supplement", "baseline")
+        for condition in order:
+            row = run_condition(condition, repetition, ready_depth)
+            row["warmup"] = False
+            results.append(row)
+            print("[L121ArrivalPair] " + json.dumps(row, sort_keys=True), flush=True)
+
+    (OUT / f"pairs-rd{ready_depth}.json").write_text(json.dumps({
+        "design": f"FI arrival, normal four applies versus one supplemental far apply, ready depth {ready_depth}",
+        "runs": results,
+    }, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

Standalone/Research/WorldAssemblerM3/Linux121/RunL121Hashes.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Compare L121 exact tile hashes with L104 across all strata, compilers and workers."""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+OUT = ROOT / "Saved/Linux/l121/hashes"
+DATA = ROOT / "Standalone/data/owner_world.json"
+OLD = ROOT / "Saved/Linux/l104/hashes"
+BINARIES = {
+    "clang": ROOT / "Saved/Linux/l121/verify/bin/VoxelForgeCoreHeadless-verify-clang",
+    "gcc": ROOT / "Saved/Linux/l121/verify/bin/VoxelForgeCoreHeadless-verify-gcc",
+}
+CAMERAS = {
+    "sw": ((8.0, 16.0, -372.2626953125), 2850),
+    "fi": ((25.8369140625, 3.19921875, -737.0), 2249),
+    "tn": ((19.166015625, -14.2001953125, -753.3740234375), 2233),
+    "vs": ((-20.0625, -5.068359375, -900.353515625), 2329),
+}
+TILE = re.compile(
+    r"^\[PipelineTileHash\] lod=(-?\d+) x=(-?\d+) y=(-?\d+) z=(-?\d+) "
+    r"density=(0x[0-9a-f]+) mesh=(0x[0-9a-f]+) payload=(0x[0-9a-f]+) "
+    r"topology=(0x[0-9a-f]+)$", re.M)
+WHOLE = re.compile(
+    r"^\[PipelineWholeViewHash\] tiles=(\d+) density=(0x[0-9a-f]+) "
+    r"mesh=(0x[0-9a-f]+) payload=(0x[0-9a-f]+) topology=(0x[0-9a-f]+)$", re.M)
+STAMP = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]{64})$", re.M)
+
+
+def parse(path: Path) -> tuple[dict, tuple[str, ...], str]:
+    content = path.read_text(encoding="utf-8", errors="replace")
+    tiles = {(int(m.group(1)), int(m.group(2)), int(m.group(3)), int(m.group(4))):
+             m.group(5, 6, 7, 8) for m in TILE.finditer(content)}
+    whole, stamp = WHOLE.search(content), STAMP.search(content)
+    if whole is None or stamp is None or len(tiles) != int(whole.group(1)):
+        raise RuntimeError(f"incomplete hash output: {path}")
+    return tiles, whole.groups(), stamp.group(1)
+
+
+def run(stratum: str, compiler: str, workers: int) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    log = OUT / f"{stratum}-{compiler}-w{workers}.log"
+    camera, expected_tiles = CAMERAS[stratum]
+    camera_arg = ",".join(f"{v:.12f}" for v in camera) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", "timeout", "1740s",
+        str(BINARIES[compiler]), str(DATA), "--world-assembler-prototype",
+        "--stream-benchmark", "--stream-pipeline", f"--workers={workers}",
+        "--clip-radius=3", "--max-clip-level=8", "--max-level=8",
+        f"--camera={camera_arg}", "--stream-hash-tiles",
+    ]
+    started = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        process = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                 stderr=subprocess.STDOUT, check=False, timeout=1740)
+    if process.returncode:
+        raise RuntimeError(f"hash process exited {process.returncode}: {log}")
+    tiles, whole, stamp = parse(log)
+    if len(tiles) != expected_tiles:
+        raise RuntimeError(f"expected {expected_tiles} tiles, got {len(tiles)}: {log}")
+    old_log = OLD / f"{stratum}-{compiler}-w{workers}.log"
+    old_tiles, old_whole, _old_stamp = parse(old_log)
+    clang_reference, clang_whole, _ = parse(OLD / f"{stratum}-clang-w1.log")
+
+    def difference(reference: dict, candidate: dict) -> tuple[int, int]:
+        keys = set(reference) | set(candidate)
+        mismatches = sum(reference.get(key) != candidate.get(key) for key in keys)
+        return mismatches, len(keys)
+
+    old_mismatches, old_keys = difference(old_tiles, tiles)
+    cross_mismatches, cross_keys = difference(clang_reference, tiles)
+    row = {
+        "stratum": stratum, "compiler": compiler, "workers": workers,
+        "source_stamp": stamp, "tiles": len(tiles),
+        "l104_same_cell_tile_mismatches": old_mismatches,
+        "l104_same_cell_key_count": old_keys,
+        "l104_same_cell_whole_hash_match": whole == old_whole,
+        "clang_w1_tile_mismatches": cross_mismatches,
+        "clang_w1_key_count": cross_keys,
+        "clang_w1_whole_hash_match": whole == clang_whole,
+        "wall_s": round(time.monotonic() - started, 3),
+        "log": str(log.relative_to(ROOT)),
+    }
+    print("[L121HashRun] " + json.dumps(row, sort_keys=True), flush=True)
+    row["tile_hashes"] = {"|".join(map(str, key)): value for key, value in tiles.items()}
+    return row
+
+
+def main() -> int:
+    for path in (*BINARIES.values(), DATA):
+        if not path.is_file():
+            raise SystemExit(f"missing input: {path}")
+    rows = []
+    for stratum in CAMERAS:
+        for compiler in BINARIES:
+            for workers in (1, 8, 12):
+                rows.append(run(stratum, compiler, workers))
+                (OUT / "runs-partial.json").write_text(
+                    json.dumps(rows, indent=2) + "\n", encoding="utf-8")
+    failures = [row for row in rows if row["l104_same_cell_tile_mismatches"]
+                or not row["l104_same_cell_whole_hash_match"]
+                or row["clang_w1_tile_mismatches"]
+                or not row["clang_w1_whole_hash_match"]]
+    summary = {"runs": [{key: value for key, value in row.items() if key != "tile_hashes"}
+                        for row in rows], "failed_runs": len(failures)}
+    (OUT / "summary.json").write_text(json.dumps(summary, indent=2) + "\n",
+                                      encoding="utf-8")
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

Standalone/Research/WorldAssemblerM3/Linux121/RunL121Movement.py

@@ -0,0 +1,285 @@
+#!/usr/bin/env python3
+"""Five interleaved L121 apply-supplement pairs at 30 m/s."""
+from __future__ import annotations
+
+import argparse
+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 = ROOT / "Saved/Linux/l108/bin/VoxelForgeCoreHeadless-l108-after-clang"
+OUT = ROOT / "Saved/Linux/l121/movement/far-apply-supplement"
+HOST_SAVED = HOST / "Saved/Linux/l121"
+HOST_OUT = HOST / "Plugins/VoxelForge/Saved/Linux/l121/far-apply-supplement"
+ACTIVE_DESIGN = "far_apply_supplement"
+SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375)
+MOVING_SAMPLE = re.compile(r"^\[L79MovingSample\] (.*)$", re.M)
+MOVING_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 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 line.startswith("{"):
+            continue
+        try:
+            events.append(json.loads(line))
+        except json.JSONDecodeError:
+            pass
+    return events
+
+
+def summarize(rows: list[dict], phase: str) -> dict:
+    moving = [row for row in rows if row["phase"] == "move"]
+    dwell = [row for row in rows if row["phase"] == "dwell"]
+    def moving_mean(name: str) -> float | None:
+        values = [row[name] for row in moving if name in row]
+        return statistics.mean(values) if values else None
+    complete = next((row for row in dwell
+                     if row["near_ready_pct"] >= 99.999
+                     and row["far_hole_pct"] <= 0.001), None)
+    end = dwell[-1] if dwell else {}
+    return {
+        "moving_samples": len(moving),
+        "dwell_samples": len(dwell),
+        "near_ready_moving_mean_pct": moving_mean("near_ready_pct"),
+        "lod0_ready_moving_mean_pct": moving_mean("lod0_ready_pct"),
+        "lod1_ready_moving_mean_pct": moving_mean("lod1_ready_pct"),
+        "lod2_ready_moving_mean_pct": moving_mean("lod2_ready_pct"),
+        "far_holes_moving_mean_pct": moving_mean("far_hole_pct"),
+        "near_ready_dwell_end_pct": end.get("near_ready_pct"),
+        "far_holes_dwell_end_pct": end.get("far_hole_pct"),
+        "dwell_complete_after_stop_s": complete.get("since_stop_s") if complete else None,
+        "phase": phase,
+    }
+
+
+def run_headless(condition: str, repetition: int) -> dict:
+    worker_count = 8
+    folder = OUT / condition / "headless"
+    folder.mkdir(parents=True, exist_ok=True)
+    log = folder / f"fast-r{repetition}.log"
+    camera = ",".join(f"{value:.12f}" for value in SPAWN_M) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "timeout", "1740s", str(HEADLESS), str(DATA),
+        "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
+        f"--workers={worker_count}", "--clip-radius=3", "--max-clip-level=8",
+        "--max-level=8", f"--camera={camera}", "--profile-lod-work",
+        "--stream-scheduler-policy=3", "--moving-camera-speed-mps=30.0",
+        "--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 MOVING_SAMPLE.finditer(content):
+        row = dict(FIELDS.findall(match.group(1)))
+        for name in ("t_s", "near_ready_pct", "walkable_ring_ready_pct",
+                     "far_hole_pct", "queue_update_ms"):
+            if name in row:
+                row[name] = float(row[name])
+        row["phase"] = row.get("phase", "move")
+        row["since_stop_s"] = max(0.0, row.get("t_s", 0.0) - 12.0)
+        rows.append(row)
+    engine_summary = {}
+    if match := MOVING_SUMMARY.search(content):
+        engine_summary = dict(FIELDS.findall(match.group(1)))
+    if result.returncode or not rows or int(engine_summary.get("generation_failures", "-1")) != 0:
+        raise RuntimeError(f"headless run failed: {log}; exit={result.returncode}")
+    moving_updates = [row["queue_update_ms"] for row in rows
+                      if row["phase"] == "move" and "queue_update_ms" in row]
+    return {
+        "condition": condition, "platform": "headless", "speed_mps": 30.0,
+        "repetition": repetition, "worker_count": worker_count,
+        "exit_code": result.returncode,
+        "wall_s": round(time.monotonic() - started, 3),
+        "update_work_p99_ms": p99(moving_updates),
+        "update_work_max_ms": max(moving_updates, default=None),
+        "log": str(log.relative_to(ROOT)), **summarize(rows, "headless"),
+    }
+
+
+def run_unreal(condition: str, repetition: int) -> dict:
+    core_limit = 12
+    worker_cap = 1
+    far_apply_supplement = 0 if condition == "baseline" else 1
+    folder = OUT / condition / "unreal"
+    host_folder = HOST_OUT / condition / "output"
+    folder.mkdir(parents=True, exist_ok=True)
+    trace = folder / f"fast-r{repetition}.trace.jsonl"
+    abslog = folder / f"fast-r{repetition}.log"
+    stdout = folder / f"fast-r{repetition}.stdout.log"
+    export = host_folder / f"fast-r{repetition}"
+    userdir = HOST_SAVED / "user" / condition / f"fast-r{repetition}"
+    zendata = HOST_SAVED / "zen" / condition / f"fast-r{repetition}"
+    ddc = HOST_SAVED / "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", f"-corelimit={core_limit}",
+        "-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",
+        f"-voxel.StreamingBackgroundWorkerCap={worker_cap}",
+        "-voxel.CoreColumnCacheDirectMapped=0",
+        f"-voxel.StreamingFarApplySupplement={far_apply_supplement}",
+        "-voxel.CoreMCGradientMemoization=1", "-voxel.StreamingCollisionLookahead=1",
+        "-voxel.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=3000.0",
+        "-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"]
+    moving_samples = [event for event in samples
+                      if float(event.get("speed_cm_s", 0.0)) >= 50.0]
+    move_start = min((float(event.get("t_s", 0.0)) for event in moving_samples),
+                     default=float("inf"))
+    stop_event = next((event for event in events
+                       if event.get("type") == "streaming_test_move_stopped"), None)
+    stop = float(stop_event.get("t_s", 0.0)) if stop_event else float("inf")
+    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)
+        near_desired = sum(desired[:3])
+        near_resident = sum(resident[:3])
+        far_desired = sum(desired[3:])
+        far_covered = sum(covered[3:])
+        t_s = float(event.get("t_s", 0.0))
+        speed = float(event.get("speed_cm_s", 0.0))
+        phase = "move" if speed >= 50.0 else ("dwell" if t_s >= stop else "arrival")
+        rows.append({
+            "phase": phase, "t_s": t_s,
+            "since_stop_s": max(0.0, t_s - stop) if phase == "dwell" else None,
+            "near_ready_pct": 100.0 * near_resident / max(1, near_desired),
+            "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]),
+            "far_hole_pct": 100.0 * (far_desired - far_covered) / max(1, far_desired),
+        })
+    dwell_rows = [row for row in rows if row["phase"] == "dwell"]
+    complete = next((row for row in dwell_rows
+                     if row["near_ready_pct"] >= 99.999
+                     and row["far_hole_pct"] <= 0.001), None)
+    moving_work = [event for event in events
+                   if event.get("type") == "game_thread_work_frame"
+                   and move_start <= float(event.get("t_s", 0.0)) < stop]
+    ticks = [float(event.get("tick_ms", 0.0)) for event in moving_work]
+    updates = [float(event.get("update_chunks_ms", 0.0)) for event in moving_work]
+    backend = next((event for event in events
+                    if event.get("type") == "worker_backend_config"), {})
+    worker_cap_readback = next((event for event in events
+                                if event.get("type") == "streaming_worker_cap_readback"), {})
+    if result.returncode or len(moving_samples) < 40 or len(dwell_rows) < 16:
+        raise RuntimeError(f"Unreal run failed/incomplete: {trace}; exit={result.returncode}; "
+                           f"move={len(moving_samples)} dwell={len(dwell_rows)}")
+    expected_far_supplement = 0 if condition == "baseline" else 1
+    expected_near_priority = "BackgroundLow"
+    if (backend.get("near_priority") != expected_near_priority
+            or backend.get("horizon_priority") != "BackgroundLow"
+            or int(worker_cap_readback.get("streaming_background_worker_cap", -1)) != worker_cap
+            or int(backend.get("streaming_far_apply_supplement", -1)) != expected_far_supplement
+            or int(backend.get("effective_tile_slots", -1)) != (8 if worker_cap else 10)):
+        raise RuntimeError(f"priority CVar readback mismatch in {trace}: {backend}")
+    return {
+        "condition": condition, "platform": "unreal", "speed_mps": 30.0,
+        "repetition": repetition, "core_limit": core_limit,
+        "effective_tile_slots": backend.get("effective_tile_slots"),
+        "streaming_background_worker_cap": worker_cap_readback.get(
+            "streaming_background_worker_cap"),
+        "taskgraph_background_threads": backend.get("taskgraph_background_threads"),
+        "exit_code": result.returncode,
+        "wall_s": round(time.monotonic() - started, 3),
+        "game_thread_tick_p99_ms": p99(ticks),
+        "game_thread_tick_max_ms": max(ticks, default=None),
+        "stream_update_p99_ms": p99(updates),
+        "stream_update_max_ms": max(updates, default=None),
+        "dwell_complete_after_stop_s": complete.get("since_stop_s") if complete else None,
+        "moving_samples": len(moving_samples), "dwell_samples": len(dwell_rows),
+        "trace": str(trace.relative_to(ROOT)),
+        "log": str(abslog.relative_to(ROOT)),
+        **summarize(rows, "unreal"),
+    }
+
+
+def main() -> int:
+    global ACTIVE_DESIGN, OUT, HOST_SAVED, HOST_OUT
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--platform", choices=("all", "headless", "unreal"), default="all")
+    parser.add_argument("--design", choices=("far_apply_supplement",),
+                        default="far_apply_supplement")
+    args = parser.parse_args()
+    ACTIVE_DESIGN = args.design
+    design_path = "far-apply-supplement"
+    OUT = ROOT / "Saved/Linux/l121/movement" / design_path
+    HOST_SAVED = HOST / "Saved/Linux/l121" / design_path
+    HOST_OUT = HOST / "Plugins/VoxelForge/Saved/Linux/l121" / design_path
+    if not DATA.is_file() or not HEADLESS.is_file() or not EDITOR.is_file():
+        raise SystemExit("missing owner data, L108 headless binary, or Unreal editor")
+    OUT.mkdir(parents=True, exist_ok=True)
+    results = []
+    partial = OUT / "runs-partial.json"
+    for repetition in range(1, 6):
+        candidate = "far_supplement"
+        conditions = ("baseline", candidate) if repetition % 2 else (candidate, "baseline")
+        platforms = (("headless", "unreal") if repetition % 2 else ("unreal", "headless")) \
+            if args.platform == "all" else (args.platform,)
+        for condition in conditions:
+            for platform in platforms:
+                row = (run_headless(condition, repetition) if platform == "headless"
+                       else run_unreal(condition, repetition))
+                results.append(row)
+                partial.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8")
+                print("[L121Movement] " + json.dumps(row, sort_keys=True), flush=True)
+    design_description = "one supplemental far result after the normal four applies, with the 2 ms time budget"
+    (OUT / "summary.json").write_text(
+        json.dumps({"design": design_description,
+                    "runs": results}, indent=2) + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())