streaming: speed-aware scheduler with hysteresis (L80); arrival near-first, fast travel keeps the horizon

Policy 3 (headless + Unreal): walkable ring always first; at rest/arrival LOD0-2 before far;
walking interleaves near:far ~4:1 toward heading; fast travel admits far coarse tiles right after
the walkable ring. Hysteresis: walking 0.75/0.35 m/s, fast 12/8 m/s.
Headless, far holes while moving 30 m/s: L77 9.0% / L79 77.2% / L80 8.9%. Cold LOD0-2 ready
(one clean protocol, medians): SW 1.30, FI 3.61, TN 3.07, VS 1.43 s.
Unreal 30 m/s traces end 2,233/2,237 loaded (commandlet readiness exit 1) for all policies.
Passage zero-padding cut: ~50% fewer passage iterations on active tiles, max 6.6 cm surface
displacement, no sign flips; not adopted yet (padding sweep next). World unchanged (v213).

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

Standalone/Research/WorldAssemblerM3/Linux80/RunL80TimingPairs.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""Clean, sampler-free, interleaved cold-readiness comparison for L77/L79/L80."""
+from __future__ import annotations
+import json
+import re
+import statistics
+import subprocess
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+OUT = ROOT / "Saved/Linux/l80/timing"
+DATA = ROOT / "Standalone/data/owner_world.json"
+BINARY = ROOT / "Saved/Linux/l80/bin/VoxelForgeCoreHeadless-l80-clean-clang"
+POLICIES = {0: "L77", 2: "L79", 3: "L80"}
+CAMERAS = {
+    0: (8.0, 16.0, -372.2626953125),
+    1: (25.8369140625, 3.19921875, -737.0),
+    2: (19.166015625, -14.2001953125, -753.3740234375),
+    3: (-20.0625, -5.068359375, -900.353515625),
+}
+PIPELINE = re.compile(r"^\[StreamPipeline\] (.*)$", re.M)
+FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
+NEAR = re.compile(r"^\[WorldAssemblerNearLODReady\] lods=0-2 tiles=(\d+)/(\d+) cold_ready_s=([0-9.]+)$", re.M)
+
+
+def run(strate: int, policy: int, phase: str, replicate: int) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    label = f"s{strate}-{POLICIES[policy]}-{phase}{replicate}"
+    log = OUT / f"{label}.log"
+    camera = ",".join(f"{v:.12f}" for v in CAMERAS[strate]) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "timeout", "1740s", str(BINARY), 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", f"--stream-scheduler-policy={policy}",
+    ]
+    start = time.monotonic()
+    with log.open("w", encoding="utf-8") as output:
+        result = subprocess.run(command, cwd=ROOT, stdout=output,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    content = log.read_text(encoding="utf-8", errors="replace")
+    pipeline_match = PIPELINE.search(content)
+    near_match = NEAR.search(content)
+    if result.returncode or pipeline_match is None or near_match is None:
+        raise SystemExit(f"readiness run failed or trace missing: {log}; exit={result.returncode}")
+    pipeline = dict(FIELDS.findall(pipeline_match.group(1)))
+    row = {
+        "strate": strate, "policy": POLICIES[policy], "policy_id": policy,
+        "phase": phase, "replicate": replicate, "exit_code": result.returncode,
+        "wall_s": round(time.monotonic() - start, 3),
+        "lod02_ready_s": float(near_match.group(3)),
+        "whole_view_ready_s": float(pipeline["whole_view_ready"].removesuffix("s")),
+        "whole_tiles": int(pipeline["whole_tiles"].split("/")[0]),
+        "log": str(log.relative_to(ROOT)),
+    }
+    print("[L80TimingRun] " + json.dumps(row, sort_keys=True), flush=True)
+    return row
+
+
+def main() -> int:
+    if not DATA.is_file() or not BINARY.is_file():
+        raise SystemExit(f"missing input: {DATA} or {BINARY}")
+    rows = []
+    # One unrecorded warm-up per stratum and policy. It warms binary/data pages only.
+    for strate in range(4):
+        for policy in (0, 2, 3):
+            run(strate, policy, "warmup", 0)
+    # Three interleaved triplets; rotate order to keep any drift from favoring one policy.
+    orders = ((0, 2, 3), (3, 2, 0), (2, 0, 3))
+    for strate in range(4):
+        for replicate, order in enumerate(orders, 1):
+            for policy in order:
+                row = run(strate, policy, "measured", replicate)
+                rows.append(row)
+                (OUT / "runs-partial.json").write_text(
+                    json.dumps(rows, indent=2) + "\n", encoding="utf-8")
+    summary = []
+    for strate in range(4):
+        group = [row for row in rows if row["strate"] == strate]
+        entry = {"strate": strate, "measured_pairs_per_policy": 3}
+        for policy in ("L77", "L79", "L80"):
+            subset = [row for row in group if row["policy"] == policy]
+            entry[f"{policy}_lod02_s"] = [row["lod02_ready_s"] for row in subset]
+            entry[f"{policy}_lod02_median_s"] = statistics.median(
+                row["lod02_ready_s"] for row in subset)
+            entry[f"{policy}_whole_s"] = [row["whole_view_ready_s"] for row in subset]
+            entry[f"{policy}_whole_median_s"] = statistics.median(
+                row["whole_view_ready_s"] for row in subset)
+        summary.append(entry)
+    (OUT / "summary.json").write_text(
+        json.dumps({"runs": rows, "summary": summary}, indent=2) + "\n",
+        encoding="utf-8")
+    return 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())