streaming: early-admission + apply-cap experiment controls, off by default; forward-cone metric (L157)

30 m/s UE, 5 pairs: near coverage 36.1 -> 48.6% but 2/5 runs had p99 15.8/18.2 ms (worst 19.9 ms), so controls stay
off. Measured: only ~23% of desired LOD0-2 tiles lie in the 55 deg forward cone; forward readiness is 12.1% (base) /
16.9%, vs 50.9/68.5% behind. Slow frames are dominated by desired-set rebuild/UpdateChunks, not apply. Source/:
VoxelWorld.cpp (default behaviour unchanged).

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

Standalone/Research/WorldAssemblerM3/Linux157/REPORT.md

@@ -0,0 +1,38 @@
+# L157 — near terrain at 30 m/s
+
+**Result:** Early admission raised moving near coverage, but the W8 experiment did not meet the full latency and worst-tick bar. Keep both experiment controls disabled by default. The UE host was built from `/home/jahni/VoxelM/Plugins/VoxelForge/Saved/Linux/l157-tree/Source`; the VFHost source symlink has since been restored.
+
+## Measurements
+
+Five interleaved pairs used the same 30 m/s, 12 s movement plus 5 s dwell, 8 effective tile slots, and four-strate owner plan. Each run had 45–46 moving samples. The uncounted candidate smoke run warmed the same UE binary. The machine was quiet before the matrix (uptime load averages 1.09/1.40/1.22).
+
+| Moving metric | `assembler/world` median | Early admission + 1 ms cap median | Paired median delta |
+| --- | ---: | ---: | ---: |
+| LOD0–2 ready | 36.06% | 48.64% | +10.76 pp |
+| 55° forward-cone ready | 12.12% | 16.86% | +5.95 pp |
+| Game-thread tick p99 | 12.51 ms | 10.93 ms | +0.07 ms |
+| Worst tick per run, median | 14.67 ms | 14.52 ms | +0.18 ms |
+| Far holes | 4.00% | 5.82% | −0.13 pp |
+| Initial near-set arrival | 2.80 s | 3.21 s | +0.01 s |
+
+The p99 median is under 11 ms, but only three candidate runs were under that limit; the other two were 15.75 and 18.17 ms. The worst observed candidate tick was 19.88 ms versus 17.75 ms for baseline. Far-hole results were mixed: the paired median improved slightly, while two candidate pairs regressed and the worst candidate run had 12.48% holes versus a 7.08% baseline worst. Arrival was effectively tied by paired median, with one candidate pair 0.88 s slower. The combined bar therefore fails on repeatability and worst tick; far holes are not robustly no-worse.
+
+The apply cap lowered median apply p99 from 2.11 to 1.40 ms, but it is not a hard per-frame ceiling: candidate apply p99 reached 6.61 ms and the maximum was 7.58 ms. The loop checks elapsed time after finishing a result. Candidate stream-update p99 reached 15.07 and 13.52 ms in the same two outlier runs. Historical L122 frame joins likewise show that applying extra results adds work on candidate tail frames (about 1.0–1.5 ms), while `UpdateChunks`/desired-set work forms the larger tail; `level_admit` and `submit_tiles` p99 were not consistently higher. The cap cannot make the whole tick fit when update work already exceeds the frame budget.
+
+## Direction and next lever
+
+The old metric counts every desired LOD0–2 tile equally, including tiles behind the player. Desired tiles split about 50.44% ahead and 49.56% behind; only 23.20% fell inside the 55° forward cone. Behind readiness was 50.93% baseline / 68.49% candidate, versus 21.71% / 30.60% ahead. The all-around count therefore overstates what is ready in the direction of travel.
+
+For VR, readiness against the HMD/camera frustum is the useful visibility metric. The velocity-forward cone measured here is a movement proxy, not a head-pose measurement. Report it beside the old all-around residency count. No additional speed fix is validated; a narrow forward-cone admission probe is the next cheap experiment, while desired-set rebuild/update work remains the measured latency target.
+
+## Checks and limits
+
+- UBT built `VFHostEditor` successfully, 98/98 actions, using `-NoUBA -MaxParallelActions=4`. It compiled `Source/VoxelForge/Private/VoxelWorld.cpp` from the L157 worktree. Three existing RealtimeMesh deprecation warnings remained.
+- Fast tier: 268 PASS, 0 FAIL, 5 older-baseline diagnostics. Current canonical owner hashes match L130; W1/W12 hashes match for both pinned owners. Random-camera tile hashes also match across W1/W8/W12.
+- UE `TileHashDump` comparison was inconclusive: both runs logged `records=0`, despite streamed meshes. No direct UE tile-hash claim is made. The L157 changes are limited to admission/apply controls and trace accounting; no generator or mesher source changed.
+- W12 movement timing and HMD-pose visibility were not run. W12 was skipped after W8 failed the worst-tick bar; direct UE hash capture emitted no records.
+- `RunL157Movement.py` passed `py_compile`; `git diff --check` passed.
+
+Logs and paired data are in `Saved/Linux/l157/`; movement traces and summary are in `Saved/Linux/l157/ue/movement/`.
+
+Changed Unreal source: `Source/VoxelForge/Private/VoxelWorld.cpp`.

Standalone/Research/WorldAssemblerM3/Linux157/RunL157Movement.py

@@ -0,0 +1,400 @@
+#!/usr/bin/env python3
+"""Interleaved L157 UE movement runs for critical-gate and apply-budget ablations."""
+from __future__ import annotations
+
+import argparse
+import json
+import math
+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")
+OUT = ROOT / "Saved/Linux/l157/ue/movement"
+HOST_SAVED = HOST / "Saved/Linux/l157"
+HOST_PLUGIN_SAVED = HOST / "Plugins/VoxelForge/Saved/Linux/l157"
+SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375)
+MOVE_START_S = 6.0
+
+CONDITIONS = {
+    "assembler_world": {"wait_for_critical": 1, "apply_budget_ms": 0.0},
+    "early_admission_apply_1ms": {"wait_for_critical": 0, "apply_budget_ms": 1.0},
+}
+ORDER = {
+    1: ("assembler_world", "early_admission_apply_1ms"),
+    2: ("early_admission_apply_1ms", "assembler_world"),
+    3: ("assembler_world", "early_admission_apply_1ms"),
+    4: ("early_admission_apply_1ms", "assembler_world"),
+    5: ("assembler_world", "early_admission_apply_1ms"),
+}
+
+
+def percentile(values: list[float], fraction: float = 0.99) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    return ordered[max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1))]
+
+
+def read_events(path: Path) -> list[dict]:
+    rows = []
+    if path.is_file():
+        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+            if not line.startswith("{"):
+                continue
+            try:
+                rows.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return rows
+
+
+def mean(values: list[float]) -> float | None:
+    return statistics.mean(values) if values else None
+
+
+def percent(numerator: int | float, denominator: int | float) -> float | None:
+    return 100.0 * numerator / denominator if denominator > 0 else None
+
+
+def run(condition: str, repetition: int, tile_hash_dump: Path | None = None) -> dict:
+    config = CONDITIONS[condition]
+    run_name = f"fast-r{repetition}"
+    out = OUT / condition
+    host_out = HOST_PLUGIN_SAVED / condition / "output"
+    userdir = HOST_SAVED / "user" / condition / run_name
+    zendata = HOST_SAVED / "zen" / condition / run_name
+    ddc = HOST_SAVED / "DDC"
+    trace = out / f"{run_name}.trace.jsonl"
+    abslog = out / f"{run_name}.log"
+    stdout = out / f"{run_name}.stdout.log"
+    export = host_out / run_name
+    for directory in (out, export, userdir, zendata, ddc):
+        directory.mkdir(parents=True, exist_ok=True)
+    trace.unlink(missing_ok=True)
+
+    spawn_cm = ",".join(f"{value * 100.0:.3f}" for value in SPAWN_M)
+    command = [
+        str(EDITOR), str(HOST / "VFHost.uproject"), "-nullrhi", "-unattended",
+        "-nop4", "-nosplash", "-NoSound", "-corelimit=12", "-NoZenAutoLaunch",
+        "-notraceserver", "-nocrashreports", "-DisablePlugins=AndroidFileServer",
+        "-ddc=NoZenLocalFallback", f"-userdir={userdir}", f"-ZenDataPath={zendata}",
+        f"-LocalDataCachePath={ddc}", "-run=VoxelForgeExplore", "-seed=0",
+        "-archetype=TunnelNetwork", "-slot=0", "-modes=export", "-gameconfig=1",
+        "-coreworldtest=1", "-ownerfourstrateplan=1", "-corevoxelsizecm=25",
+        "-coreislandtest=0", "-corefallbacktest=0", "-coreofftest=0",
+        "-coreplantest=0", "-settingsref=/Game/VoxelForge/DA_Settings.DA_Settings",
+        "-corehybridmesher=0", f"-out={export}", "-voxel.StreamingCoarseFirst=1",
+        "-voxel.StreamingDynamicNearReserve=1",
+        "-voxel.StreamingLegacyPendingResultCap=0",
+        "-voxel.CoreRegionConcurrentAdmission=1",
+        "-voxel.StreamingResultBacklogCap=256", "-voxel.StreamingReadyQueueDepth=8",
+        "-voxel.StreamingBackgroundWorkerCap=1", "-voxel.CoreColumnCacheDirectMapped=0",
+        "-voxel.StreamingFarApplySupplement=1",
+        f"-voxel.StreamingFastTravelWaitForAllCriticalTiles={config['wait_for_critical']}",
+        f"-voxel.StreamingFastTravelApplyBudgetMilliseconds={config['apply_budget_ms']:.1f}",
+        "-voxel.CoreMCGradientMemoization=1", "-voxel.StreamingCollisionLookahead=1",
+        "-voxel.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=3000.0",
+        f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}",
+        "-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}",
+    ]
+    if tile_hash_dump is not None:
+        tile_hash_dump.parent.mkdir(parents=True, exist_ok=True)
+        tile_hash_dump.unlink(missing_ok=True)
+        command.append(f"-voxel.TileHashDump={tile_hash_dump.resolve()}")
+    wrapped = ["taskset", "-c", "0-11", "timeout", "1740s", *command]
+    started = time.monotonic()
+    with stdout.open("w", encoding="utf-8") as stream:
+        process = subprocess.run(
+            wrapped, cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT,
+            check=False, timeout=1740)
+
+    events = read_events(trace)
+    samples = [event for event in events if event.get("type") == "streaming_sample"]
+    moving = [event for event in samples if float(event.get("speed_cm_s", 0.0)) >= 50.0]
+    stop_event = next((event for event in events
+                       if event.get("type") == "streaming_test_move_stopped"), None)
+    backend = next((event for event in events if event.get("type") == "worker_backend_config"), {})
+    worker_cap = next((event for event in events
+                       if event.get("type") == "streaming_worker_cap_readback"), {})
+    if stop_event is None:
+        raise RuntimeError(f"missing move-stop event: {trace}")
+    stop_s = float(stop_event.get("t_s", 0.0))
+    move_start_s = min((float(event.get("t_s", 0.0)) for event in moving), default=float("inf"))
+    dwell = [event for event in samples if float(event.get("t_s", 0.0)) >= stop_s
+             and float(event.get("speed_cm_s", 0.0)) < 50.0]
+    arrival = [event for event in samples if float(event.get("t_s", 0.0)) < move_start_s
+               and float(event.get("speed_cm_s", 0.0)) < 50.0]
+    if process.returncode or len(moving) < 40 or len(dwell) < 16:
+        raise RuntimeError(f"incomplete {condition} run: {trace}; exit={process.returncode}; "
+                           f"move={len(moving)} dwell={len(dwell)}")
+    if (int(backend.get("streaming_fast_travel_wait_for_all_critical_tiles", -1))
+            != config["wait_for_critical"]
+            or abs(float(backend.get("streaming_fast_travel_apply_budget_ms", -1.0))
+                   - config["apply_budget_ms"]) > 0.01
+            or int(worker_cap.get("streaming_background_worker_cap", -1)) != 1
+            or int(backend.get("effective_tile_slots", -1)) != 8
+            or int(backend.get("streaming_far_apply_supplement", -1)) != 1):
+        raise RuntimeError(f"CVar/worker readback mismatch: {backend}; {worker_cap}")
+
+    moving_metrics = []
+    for event in moving:
+        desired = event.get("desired_by_lod", [0] * 9)
+        resident = event.get("resident_by_lod", [0] * 9)
+        near_desired = sum(desired[:3])
+        cone_desired = int(event.get("near_forward_cone_desired", 0))
+        weighted_desired = float(event.get("near_forward_weighted_desired", 0.0))
+        moving_metrics.append({
+            "near_ready_pct": percent(sum(resident[:3]), near_desired) or 0.0,
+            "lod0_ready_pct": percent(resident[0], desired[0]) or 0.0,
+            "lod1_ready_pct": percent(resident[1], desired[1]) or 0.0,
+            "lod2_ready_pct": percent(resident[2], desired[2]) or 0.0,
+            "far_hole_pct": percent(
+                sum(desired[3:]) - sum(event.get("covered_by_lod", resident)[3:]),
+                sum(desired[3:])) or 0.0,
+            "ahead_share_pct": percent(int(event.get("near_ahead_desired", 0)), near_desired),
+            "behind_share_pct": percent(int(event.get("near_behind_desired", 0)), near_desired),
+            "side_share_pct": 100.0 - (
+                percent(int(event.get("near_ahead_desired", 0)), near_desired) or 0.0
+            ) - (
+                percent(int(event.get("near_behind_desired", 0)), near_desired) or 0.0
+            ),
+            "ahead_ready_pct": percent(
+                int(event.get("near_ahead_resident", 0)),
+                int(event.get("near_ahead_desired", 0))),
+            "behind_ready_pct": percent(
+                int(event.get("near_behind_resident", 0)),
+                int(event.get("near_behind_desired", 0))),
+            "forward_cone_ready_pct": percent(
+                int(event.get("near_forward_cone_resident", 0)), cone_desired),
+            "forward_cone_share_pct": percent(cone_desired, near_desired),
+            "forward_weighted_ready_pct": percent(
+                float(event.get("near_forward_weighted_resident", 0.0)), weighted_desired),
+        })
+
+    arrival_near_settled = next((event for event in arrival
+        if sum(event.get("desired_by_lod", [0] * 9)[:3]) > 0
+        and sum(event.get("settled_by_lod", [0] * 9)[:3])
+            >= sum(event.get("desired_by_lod", [0] * 9)[:3])), None)
+    dwell_complete = next((event for event in dwell
+        if sum(event.get("resident_by_lod", [0] * 9)[:3])
+            >= sum(event.get("desired_by_lod", [0] * 9)[:3])
+        and sum(event.get("desired_by_lod", [0] * 9)[3:])
+            <= sum(event.get("covered_by_lod", [0] * 9)[3:])), None)
+
+    work = [event for event in events if event.get("type") == "game_thread_work_frame"
+            and move_start_s <= float(event.get("t_s", 0.0)) < stop_s]
+    applies = {event.get("frame"): event for event in events
+               if event.get("type") == "apply_frame"
+               and move_start_s <= float(event.get("t_s", 0.0)) < stop_s}
+    scheduler = {event.get("frame"): event for event in events
+                 if event.get("type") == "scheduler_admission"
+                 and move_start_s <= float(event.get("t_s", 0.0)) < stop_s}
+    frames = []
+    for event in work:
+        apply = applies.get(event.get("frame"), {})
+        admit = scheduler.get(event.get("frame"), {})
+        frames.append({
+            "frame": event.get("frame"),
+            "tick_ms": float(event.get("tick_ms", 0.0)),
+            "update_chunks_ms": float(event.get("update_chunks_ms", 0.0)),
+            "pending_drain_ms": float(event.get("pending_drain_ms", 0.0)),
+            "apply_ms": float(apply.get("apply_ms", 0.0)),
+            "apply_budget_ms": float(apply.get("apply_budget_ms", 0.0)),
+            "meshes_applied": int(apply.get("meshes_applied", 0)),
+            "time_budget_hit": int(apply.get("time_limit_reached", 0)),
+            "desired_rebuild_ms": float(admit.get("desired_rebuild_ms", 0.0)),
+            "submit_tiles_ms": float(admit.get("submit_tiles_ms", 0.0)),
+            "level_admit_ms": float(admit.get("level_admit_ms", 0.0)),
+        })
+    p99_field = lambda key: percentile([float(row[key]) for row in frames])
+    max_field = lambda key: max((float(row[key]) for row in frames), default=None)
+    dwell_complete_s = (float(dwell_complete.get("t_s", 0.0)) - stop_s
+                        if dwell_complete else None)
+
+    row = {
+        "condition": condition,
+        "repetition": repetition,
+        "speed_mps": 30.0,
+        "wait_for_all_critical_tiles": config["wait_for_critical"],
+        "fast_travel_apply_budget_ms": config["apply_budget_ms"],
+        "effective_tile_slots": backend.get("effective_tile_slots"),
+        "exit_code": process.returncode,
+        "wall_s": round(time.monotonic() - started, 3),
+        "moving_samples": len(moving),
+        "dwell_samples": len(dwell),
+        "arrival_samples": len(arrival),
+        "near_ready_moving_mean_pct": mean([m["near_ready_pct"] for m in moving_metrics]),
+        "lod0_ready_moving_mean_pct": mean([m["lod0_ready_pct"] for m in moving_metrics]),
+        "lod1_ready_moving_mean_pct": mean([m["lod1_ready_pct"] for m in moving_metrics]),
+        "lod2_ready_moving_mean_pct": mean([m["lod2_ready_pct"] for m in moving_metrics]),
+        "far_holes_moving_mean_pct": mean([m["far_hole_pct"] for m in moving_metrics]),
+        "near_ahead_desired_share_pct": mean([m["ahead_share_pct"] for m in moving_metrics
+                                               if m["ahead_share_pct"] is not None]),
+        "near_behind_desired_share_pct": mean([m["behind_share_pct"] for m in moving_metrics
+                                                if m["behind_share_pct"] is not None]),
+        "near_side_desired_share_pct": mean([m["side_share_pct"] for m in moving_metrics]),
+        "near_ahead_ready_pct": mean([m["ahead_ready_pct"] for m in moving_metrics
+                                       if m["ahead_ready_pct"] is not None]),
+        "near_behind_ready_pct": mean([m["behind_ready_pct"] for m in moving_metrics
+                                        if m["behind_ready_pct"] is not None]),
+        "near_forward_cone_half_angle_deg": 55,
+        "near_forward_cone_desired_share_pct": mean([m["forward_cone_share_pct"]
+                                                      for m in moving_metrics
+                                                      if m["forward_cone_share_pct"] is not None]),
+        "near_forward_cone_ready_pct": mean([m["forward_cone_ready_pct"]
+                                              for m in moving_metrics
+                                              if m["forward_cone_ready_pct"] is not None]),
+        "near_forward_cosine_weighted_ready_pct": mean(
+            [m["forward_weighted_ready_pct"] for m in moving_metrics
+             if m["forward_weighted_ready_pct"] is not None]),
+        "arrival_lod02_settled_t_s": (
+            float(arrival_near_settled.get("t_s", 0.0)) if arrival_near_settled else None),
+        "near_ready_dwell_end_pct": (100.0 * sum(dwell[-1].get("resident_by_lod", [0] * 9)[:3])
+                                      / max(1, sum(dwell[-1].get("desired_by_lod", [0] * 9)[:3]))),
+        "far_holes_dwell_end_pct": (100.0 * (
+            sum(dwell[-1].get("desired_by_lod", [0] * 9)[3:])
+            - sum(dwell[-1].get("covered_by_lod", [0] * 9)[3:]))
+            / max(1, sum(dwell[-1].get("desired_by_lod", [0] * 9)[3:]))),
+        "dwell_complete_after_stop_s": dwell_complete_s,
+        "game_thread_tick_p99_ms": p99_field("tick_ms"),
+        "game_thread_tick_max_ms": max_field("tick_ms"),
+        "stream_update_p99_ms": p99_field("update_chunks_ms"),
+        "pending_drain_p99_ms": p99_field("pending_drain_ms"),
+        "apply_p99_ms": p99_field("apply_ms"),
+        "apply_max_ms": max_field("apply_ms"),
+        "apply_budget_min_ms": min((f["apply_budget_ms"] for f in frames), default=None),
+        "apply_budget_max_ms": max((f["apply_budget_ms"] for f in frames), default=None),
+        "apply_time_budget_hit_frames": sum(f["time_budget_hit"] for f in frames),
+        "apply_frames": sum(f["meshes_applied"] > 0 for f in frames),
+        "apply_meshes_total": sum(f["meshes_applied"] for f in frames),
+        "desired_rebuild_p99_ms": p99_field("desired_rebuild_ms"),
+        "submit_tiles_p99_ms": p99_field("submit_tiles_ms"),
+        "level_admit_p99_ms": p99_field("level_admit_ms"),
+        "top_tick_frames": sorted(frames, key=lambda f: f["tick_ms"], reverse=True)[:5],
+        "trace": str(trace.relative_to(ROOT)),
+        "log": str(abslog.relative_to(ROOT)),
+    }
+    print("[L157Movement] " + json.dumps(row, sort_keys=True), flush=True)
+    return row
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--smoke", action="store_true",
+                        help="run one candidate trace to validate CVar readback and event fields")
+    parser.add_argument("--geometry-hash-check", action="store_true",
+                        help="compare UE per-tile geometry hashes for baseline and early admission")
+    args = parser.parse_args()
+    missing = [path for path in (EDITOR, HOST / "VFHost.uproject") if not path.exists()]
+    if missing:
+        raise SystemExit(f"missing UE runner input: {missing}")
+    OUT.mkdir(parents=True, exist_ok=True)
+    if args.geometry_hash_check:
+        hash_dir = OUT / "geometry-hash"
+        baseline_path = hash_dir / "assembler-world.json"
+        candidate_path = hash_dir / "early-admission-apply-1ms.json"
+        run("assembler_world", 99, baseline_path)
+        run("early_admission_apply_1ms", 99, candidate_path)
+        baseline_data = json.loads(baseline_path.read_text(encoding="utf-8"))
+        candidate_data = json.loads(candidate_path.read_text(encoding="utf-8"))
+        key_fields = ("level", "tile_x", "tile_y", "tile_z")
+
+        def indexed(data: dict) -> dict[tuple[int, ...], tuple]:
+            result = {}
+            for record in data.get("records", []):
+                key = tuple(int(record[field]) for field in key_fields)
+                result[key] = (str(record.get("geometry_hash", "")),
+                               int(record.get("triangle_count", -1)),
+                               bool(record.get("empty", False)))
+            return result
+
+        baseline_hashes = indexed(baseline_data)
+        candidate_hashes = indexed(candidate_data)
+        common = baseline_hashes.keys() & candidate_hashes.keys()
+        mismatches = [key for key in common
+                      if baseline_hashes[key] != candidate_hashes[key]]
+        result = {
+            "baseline_records": len(baseline_hashes),
+            "candidate_records": len(candidate_hashes),
+            "common_records": len(common),
+            "hash_mismatches": len(mismatches),
+            "baseline_only_records": len(baseline_hashes.keys() - candidate_hashes.keys()),
+            "candidate_only_records": len(candidate_hashes.keys() - baseline_hashes.keys()),
+            "baseline_dump": str(baseline_path.relative_to(ROOT)),
+            "candidate_dump": str(candidate_path.relative_to(ROOT)),
+        }
+        (hash_dir / "comparison.json").write_text(
+            json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        print("[L157GeometryHash] " + json.dumps(result, sort_keys=True), flush=True)
+        return 0 if len(common) >= 100 and not mismatches else 1
+    if args.smoke:
+        run("early_admission_apply_1ms", 0)
+        return 0
+    rows = []
+    partial = OUT / "runs-partial.json"
+    for repetition in range(1, 6):
+        for condition in ORDER[repetition]:
+            row = run(condition, repetition)
+            rows.append(row)
+            partial.write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n",
+                               encoding="utf-8")
+
+    metrics = (
+        "near_ready_moving_mean_pct", "far_holes_moving_mean_pct",
+        "game_thread_tick_p99_ms", "game_thread_tick_max_ms",
+        "arrival_lod02_settled_t_s", "dwell_complete_after_stop_s",
+        "near_ahead_desired_share_pct", "near_behind_desired_share_pct",
+        "near_side_desired_share_pct", "near_forward_cone_desired_share_pct",
+        "near_ahead_ready_pct", "near_behind_ready_pct",
+        "near_forward_cone_ready_pct", "near_forward_cosine_weighted_ready_pct",
+        "apply_p99_ms", "desired_rebuild_p99_ms", "level_admit_p99_ms",
+    )
+    medians = {
+        condition: {metric: statistics.median(
+            row[metric] for row in rows if row["condition"] == condition)
+            for metric in metrics}
+        for condition in CONDITIONS
+    }
+    pairs = []
+    for repetition in range(1, 6):
+        by_condition = {row["condition"]: row for row in rows
+                        if row["repetition"] == repetition}
+        for candidate in ("early_admission_apply_1ms",):
+            pairs.append({
+                "repetition": repetition,
+                "candidate": candidate,
+                **{f"{metric}_delta": by_condition[candidate][metric]
+                   - by_condition["assembler_world"][metric]
+                   for metric in metrics if by_condition[candidate][metric] is not None
+                   and by_condition["assembler_world"][metric] is not None},
+            })
+    summary = {
+        "source_commit": subprocess.run(
+            ["git", "-c", f"safe.directory={ROOT}", "rev-parse", "HEAD"],
+            cwd=ROOT, check=True, capture_output=True, text=True).stdout.strip(),
+        "protocol": "Five interleaved 30 m/s UE runs per condition; 12 s move plus 5 s dwell; 8 effective tile slots.",
+        "conditions": CONDITIONS,
+        "movement_medians": medians,
+        "paired_deltas_vs_assembler_world": pairs,
+        "runs": rows,
+    }
+    (OUT / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n",
+                                       encoding="utf-8")
+    print("[L157Summary] " + json.dumps({"movement_medians": medians,
+                                         "runs": len(rows)}, sort_keys=True), flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())