streaming: far LODs keep a share while near work is pending (L79); no more horizon starvation

L78's strict near-first order starved far LODs at 30 m/s (100% far holes, never completed).
Unreal policy 2: once the collision ring is ready, LOD0-2 keep their reserve and far tiles are
admitted as low-priority Horizon work; headless gives far work 1 of every 16 requests.
Moving camera, far holes while moving: 1.4 m/s L77 8% / L78 29% / L79 31%; 30 m/s L77 8% /
L78 100% / L79 72% (completes on dwell). Near readiness while moving at 30 m/s: 52 / 51 / 67%.
Still worse than L77 for fast travel -> speed-adaptive policy next. World unchanged (v213).
Fast tier PASS; Linux UE host builds.

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

Standalone/Research/WorldAssemblerM3/Linux79/RunL79MovingCamera.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""Compare L77, L78, and L79 production streaming orders while the pawn moves."""
+from __future__ import annotations
+
+import csv
+import json
+import re
+import subprocess
+import time
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[4]
+HOST = Path("/home/jahni/VFHost")
+PLUGIN_SAVED = ROOT / "Saved/Linux/l79"
+HOST_PLUGIN_SAVED = HOST / "Plugins/VoxelForge/Saved/Linux/l79"
+HOST_SAVED = HOST / "Saved/Linux/l79"
+EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd")
+SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375)  # owner TunnelNetwork
+SPEEDS = {"walk": 140.0, "fast": 3000.0}  # cm/s: 1.4 m/s and 30 m/s
+POLICY_NAMES = {0: "L77", 1: "L78", 2: "L79"}
+TRACE_DIR = PLUGIN_SAVED / "movement/traces"
+MOVE_START_S = 6.0
+EXIT_S = 18.0
+
+
+def parse_trace(path: Path, run: dict) -> tuple[list[dict], dict]:
+    events = []
+    if path.is_file():
+        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+            try:
+                events.append(json.loads(line))
+            except json.JSONDecodeError:
+                pass
+    samples = [event for event in events if event.get("type") == "streaming_sample"
+               and float(event.get("speed_cm_s", 0.0)) >= 50.0]
+    scheduler = [event for event in events if event.get("type") == "scheduler_admission"
+                 and float(event.get("moving", 0)) > 0]
+    rows = []
+    for event in samples:
+        desired = event.get("desired_by_lod", [0] * 9)
+        resident = event.get("resident_by_lod", [0] * 9)
+        pending = event.get("pending_by_lod", [0] * 9)
+        far_desired = sum(desired[3:])
+        far_resident = sum(resident[3:])
+        near_desired = sum(desired[:3])
+        near_resident = sum(resident[:3])
+        rows.append({
+            "run": run["label"], "policy": run["policy_name"], "speed": run["speed"],
+            "t_s": event.get("t_s"), "position_cm": event.get("position_cm"),
+            "speed_cm_s": event.get("speed_cm_s"),
+            "far_desired": far_desired, "far_resident": far_resident,
+            "far_pending": sum(pending[3:]),
+            "far_missing_pct": round(100.0 * (far_desired - far_resident)
+                                      / max(1, far_desired), 3),
+            "near_desired": near_desired, "near_resident": near_resident,
+            "near_ready_pct": round(100.0 * near_resident / max(1, near_desired), 3),
+            "walkable_ready": event.get("walkable_ready"),
+            **{f"lod{i}_desired": desired[i] for i in range(9)},
+            **{f"lod{i}_resident": resident[i] for i in range(9)},
+        })
+    submitted_far = sum(sum(event.get("submitted_by_lod", [0] * 9)[3:])
+                        for event in scheduler)
+    submitted_near = sum(sum(event.get("submitted_by_lod", [0] * 9)[:3])
+                         for event in scheduler)
+    far_missing = [row["far_missing_pct"] for row in rows]
+    near_ready = [row["near_ready_pct"] for row in rows]
+    abslog_lines = (run["abslog_path"].read_text(encoding="utf-8", errors="replace").splitlines()
+                    if run["abslog_path"].is_file() else [])
+    summary = {
+        "moving_samples": len(rows),
+        "moving_duration_s": round(max((float(row["t_s"]) for row in rows), default=0.0)
+                                    - min((float(row["t_s"]) for row in rows), default=0.0), 3),
+        "far_missing_pct_mean": round(sum(far_missing) / max(1, len(far_missing)), 3),
+        "far_missing_pct_max": max(far_missing, default=None),
+        "far_missing_pct_first": far_missing[0] if far_missing else None,
+        "far_missing_pct_last": far_missing[-1] if far_missing else None,
+        "near_ready_pct_mean": round(sum(near_ready) / max(1, len(near_ready)), 3),
+        "near_ready_pct_min": min(near_ready, default=None),
+        "near_ready_pct_last": near_ready[-1] if near_ready else None,
+        "walkable_ready_samples": sum(row["walkable_ready"] == 1 for row in rows),
+        "submitted_near_tiles": submitted_near,
+        "submitted_far_tiles": submitted_far,
+        "lod0_ring_ready_events": sum(event.get("type") == "lod0_ring_ready" for event in events),
+        "visual_gap_result": next((line.split("result=")[1].split()[0]
+            for line in abslog_lines
+            if "[VoxelForgeVisualGapAudit] result=" in line), "not-run"),
+        "visual_staleness_result": next((line.split("result=")[1].split()[0]
+            for line in abslog_lines
+            if "[VoxelForgeVisualStaleness] result=" in line), "not-run"),
+    }
+    return rows, summary
+
+
+def run(policy: int, speed_name: str, replicate: int, audit: bool = False) -> tuple[dict, list[dict]]:
+    label = f"{speed_name}-{POLICY_NAMES[policy]}-r{replicate}" + ("-audit" if audit else "")
+    trace = TRACE_DIR / f"{label}.trace.jsonl"
+    abslog = TRACE_DIR / f"{label}.log"
+    stdout = TRACE_DIR / 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 (TRACE_DIR, output, userdir, zendata, ddc):
+        directory.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={output}", "-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={SPEEDS[speed_name]:.1f}",
+        f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}",
+        f"-voxel.TestExitSeconds={EXIT_S:.1f}", "-voxel.TestMoveDirection=1,0,0",
+        f"-voxel.StreamingSchedulerPolicy={policy}", "-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 audit:
+        command.extend(("-voxel.TestVisualStaleness=2", "-voxel.TestVisualGapAudit=1"))
+    wrapped = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+               "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)
+    run_info = {
+        "label": label, "policy": policy, "policy_name": POLICY_NAMES[policy],
+        "speed": speed_name, "speed_cm_s": SPEEDS[speed_name], "replicate": replicate,
+        "audit": audit, "exit_code": result.returncode,
+        "elapsed_s": round(time.monotonic() - started, 3),
+        "trace": str(trace.relative_to(ROOT)), "abslog": str(abslog.relative_to(ROOT)),
+        "stdout": str(stdout.relative_to(ROOT)), "abslog_path": abslog,
+    }
+    samples, summary = parse_trace(trace, run_info)
+    run_info.update(summary)
+    del run_info["abslog_path"]
+    print("[L79MoveRun] " + json.dumps(run_info, sort_keys=True), flush=True)
+    return run_info, samples
+
+
+def main() -> int:
+    if not EDITOR.is_file():
+        raise SystemExit(f"missing Unreal commandlet executable: {EDITOR}")
+    TRACE_DIR.mkdir(parents=True, exist_ok=True)
+    results: list[dict] = []
+    sample_rows: list[dict] = []
+    schedule = []
+    for replicate, policies in ((1, (0, 1, 2)), (2, (2, 1, 0))):
+        speeds = ("walk", "fast") if replicate == 1 else ("fast", "walk")
+        for policy in policies:
+            for speed in speeds:
+                schedule.append((policy, speed, replicate, False))
+    schedule.append((2, "fast", 3, True))
+    for policy, speed, replicate, audit in schedule:
+        row, samples = run(policy, speed, replicate, audit)
+        results.append(row)
+        sample_rows.extend(samples)
+        (PLUGIN_SAVED / "movement/runs-partial.json").write_text(
+            json.dumps(results, indent=2) + "\n", encoding="utf-8")
+    (PLUGIN_SAVED / "movement/summary.json").write_text(
+        json.dumps({"runs": results}, indent=2) + "\n", encoding="utf-8")
+    sample_path = PLUGIN_SAVED / "movement/samples.csv"
+    if sample_rows:
+        with sample_path.open("w", newline="", encoding="utf-8") as stream:
+            writer = csv.DictWriter(stream, fieldnames=list(sample_rows[0]))
+            writer.writeheader()
+            writer.writerows(sample_rows)
+    return 0 if all(row["exit_code"] == 0 and row["moving_samples"] > 0
+                    for row in results) else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())