research: arrival/far streaming study (L120, no production change)

Far holes coexist with idle workers: far tiles wait median 1.12 s / p95 3.21 s from worker completion
to apply while generating in p95 10 ms, so the game-thread result queue and admission are the limit.
Priority split failed; 10 slots helped but broke the worst-tick bar. LOD2 still waits for far parents.

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

Standalone/Research/WorldAssemblerM3/Linux120/RunL120ArrivalPairs.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""Five interleaved FI arrival pairs, comparing 8 and 10 effective tile slots."""
+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("RunL120Arrival.py")
+SPEC = importlib.util.spec_from_file_location("l120_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 if condition == "baseline" else 0
+    expected_slots = 8 if condition == "baseline" else 10
+    row = ARRIVAL.run(
+        "FI", repetition, condition, ready_depth, cap, 0, 1, 1, 0, 1, False, False)
+    row = add_view_metrics(row)
+    if row["effective_tile_slots"] != expected_slots:
+        raise RuntimeError(f"worker slot readback mismatch: {row}")
+    if row["streaming_tile_priority_split"] != 0:
+        raise RuntimeError(f"priority condition 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/l120/arrival"
+    OUT.mkdir(parents=True, exist_ok=True)
+    results = []
+    for condition in ("baseline", "slots10"):
+        warmup = run_condition(condition, 0, ready_depth)
+        warmup["warmup"] = True
+        results.append(warmup)
+
+    for repetition in range(1, 6):
+        order = ("baseline", "slots10") if repetition % 2 else ("slots10", "baseline")
+        for condition in order:
+            row = run_condition(condition, repetition, ready_depth)
+            row["warmup"] = False
+            results.append(row)
+            print("[L120ArrivalPair] " + json.dumps(row, sort_keys=True), flush=True)
+
+    (OUT / f"pairs-rd{ready_depth}.json").write_text(json.dumps({
+        "design": f"FI arrival, 8 versus 10 effective tile slots, priority split off, ready depth {ready_depth}",
+        "runs": results,
+    }, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())