unreal: deeper ready queue keeps the 8 tile workers fed (L95); SW near set 2.37 -> 1.31 s

Unreal per-tile timeline: per-tile Core work is not slower than headless (FI 11.2 vs 16.2 ms
per shared key); the gap was tick-paced admission with a small queue (FI ~5.2 of 8 slots busy).
voxel.StreamingReadyQueueDepth default 16 -> 64 (effective 32 with 8 slots, total cap 40);
result backlog cap 256 unchanged. First admission -> LOD0-2 settled: FI 2.72 -> 2.10 s,
SW 2.37 -> 1.31 s (3 of 4 runs; one 6.9 s outlier re-admitting 3 LOD0 keys, cause open).
apply_ms max 4.2 ms (14 ms budget). Frame-interval max 118-145 ms both before and after (open).
World unchanged. 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

Source/VoxelForge/Private/VoxelWorld.cpp

@@ -248,7 +248,7 @@ namespace
         GVoxelForgeStreamingResultBacklogCap,
         TEXT("Maximum queued results plus admitted tile tasks; 0 disables the safety cap. This is separate from the worker-slot cap."));
 
-    int32 GVoxelForgeStreamingReadyQueueDepth = 16;
+    int32 GVoxelForgeStreamingReadyQueueDepth = 64;
     FAutoConsoleVariableRef CVarVoxelForgeStreamingReadyQueueDepth(
         TEXT("voxel.StreamingReadyQueueDepth"),
         GVoxelForgeStreamingReadyQueueDepth,
@@ -7658,6 +7658,31 @@ bool AVoxelWorld::InitializeVoxelForgeCoreWorld(FString& OutFailureReason)
     const bool bSnapshotSerialized = VoxelForgeCoreAdapter::SerializeResolvedWorld(
             Settings, StrateManager, ActiveCoreClipRadius, ActiveCoreMaxClipLevel,
             ActiveCoreVoxelSizeCm, Json, ExportError);
+    FString CoreSnapshotOutputPath;
+    if (bSnapshotSerialized
+        && FParse::Value(FCommandLine::Get(),
+            TEXT("voxel.StartupTraceCoreSnapshotFile="), CoreSnapshotOutputPath))
+    {
+        CoreSnapshotOutputPath.TrimQuotesInline();
+        if (!CoreSnapshotOutputPath.IsEmpty())
+        {
+            if (FFileHelper::SaveStringToFile(Json, *CoreSnapshotOutputPath,
+                    FFileHelper::EEncodingOptions::ForceUTF8WithoutBOM))
+            {
+                VoxelForgeStartupTrace::RecordEvent(TEXT("core_snapshot_export"), FString::Printf(
+                    TEXT("\"path\":\"%s\",\"json_chars\":%d"),
+                    *CoreSnapshotOutputPath.ReplaceCharWithEscapedChar(), Json.Len()));
+            }
+            else
+            {
+                UE_LOG(LogTemp, Error,
+                    TEXT("[VoxelForgeCoreRuntime] Could not write diagnostic Core snapshot to %s"),
+                    *CoreSnapshotOutputPath);
+                OutFailureReason = TEXT("diagnostic Core snapshot export failed");
+                return false;
+            }
+        }
+    }
     if (bTraceCoreStartup)
     {
         const double SnapshotSerializeEndSeconds = FPlatformTime::Seconds();

Standalone/Research/WorldAssemblerM3/Linux95/AnalyzeL95Timelines.py

@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+"""Export per-tile and per-frame L95 Unreal streaming timelines."""
+from __future__ import annotations
+
+import csv
+import json
+import re
+import statistics
+from collections import defaultdict
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+SAVED = ROOT / "Saved/Linux/l95"
+BEFORE = SAVED / "before/unreal-arrival"
+AFTER = SAVED / "unreal-arrival"
+HEADLESS = SAVED / "headless"
+OUT = SAVED / "timelines"
+MOVE_START_S = 20.0
+HEADLESS_EVENT = re.compile(r"^\[CpuTileTimeline\] event=(\S+) (.*)$", re.M)
+HEADLESS_FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
+HEADLESS_KEY = re.compile(r"key=\((-?\d+),(-?\d+),(-?\d+),lod(-?\d+)\)")
+
+
+def read_jsonl(path: Path) -> list[dict]:
+    rows = []
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        if line.startswith("{"):
+            try:
+                rows.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return rows
+
+
+def sum3(event: dict, field: str) -> int:
+    return sum(event.get(field, [0] * 9)[:3])
+
+
+def first_settled_sample(events: list[dict]) -> dict:
+    for event in events:
+        if event.get("type") != "streaming_sample" or event.get("t_s", 0) >= MOVE_START_S:
+            continue
+        desired = sum3(event, "desired_by_lod")
+        if desired and sum3(event, "settled_by_lod") >= desired:
+            return event
+    raise RuntimeError("no initial LOD0-2 settled sample")
+
+
+def write_csv(path: Path, rows: list[dict]) -> None:
+    if not rows:
+        return
+    fields = list(rows[0])
+    with path.open("w", newline="", encoding="utf-8") as stream:
+        writer = csv.DictWriter(stream, fieldnames=fields)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def headless_tiles(path: Path) -> dict[tuple[int, int, int, int], dict]:
+    content = path.read_text(encoding="utf-8", errors="replace")
+    tiles: dict[tuple[int, int, int, int], dict] = {}
+    for match in HEADLESS_EVENT.finditer(content):
+        event, payload = match.groups()
+        if event != "finished":
+            continue
+        fields = dict(HEADLESS_FIELDS.findall(payload))
+        key_match = HEADLESS_KEY.search(payload)
+        if not key_match:
+            continue
+        x, y, z, lod = (int(value) for value in key_match.groups())
+        if lod > 2:
+            continue
+        key = (x, y, z, lod)
+        tiles[key] = {"worker_id": int(fields["worker_id"]),
+                      "worker_ms": float(fields["worker_ms"]),
+                      "queued_ms": float(fields["queued_ms"]),
+                      "dispatch_ms": float(fields["dispatch_ms"]),
+                      "finish_ms": float(fields["finish_ms"])}
+    return tiles
+
+
+def headless_lifecycle(path: Path) -> list[dict]:
+    content = path.read_text(encoding="utf-8", errors="replace")
+    tiles: dict[tuple[int, int, int, int], dict] = defaultdict(dict)
+    for match in HEADLESS_EVENT.finditer(content):
+        event, payload = match.groups()
+        fields = dict(HEADLESS_FIELDS.findall(payload))
+        key_match = HEADLESS_KEY.search(payload)
+        if not key_match:
+            continue
+        x, y, z, lod = (int(value) for value in key_match.groups())
+        if lod > 2 or event not in {
+            "requested", "admitted", "started", "finished", "published", "drained"
+        }:
+            continue
+        key = (x, y, z, lod)
+        tiles[key][event] = fields
+    rows = []
+    for key, events in sorted(tiles.items(), key=lambda item: (item[0][3], item[0][2], item[0][1], item[0][0])):
+        at = {name: (float(events[name]["at_ms"]) if name in events else None)
+              for name in ("requested", "admitted", "started", "finished", "published", "drained")}
+        started = events.get("started", {})
+        finished = events.get("finished", {})
+        rows.append({
+            "run": path.stem, "x": key[0], "y": key[1], "z": key[2], "lod": key[3],
+            **{f"{name}_ms": value for name, value in at.items()},
+            "request_to_admit_ms": round(at["admitted"] - at["requested"], 3)
+                if at["admitted"] is not None and at["requested"] is not None else None,
+            "admit_to_start_ms": round(at["started"] - at["admitted"], 3)
+                if at["started"] is not None and at["admitted"] is not None else None,
+            "worker_service_ms": float(finished.get("worker_ms", "nan")),
+            "finish_to_publish_ms": round(at["published"] - at["finished"], 3)
+                if at["published"] is not None and at["finished"] is not None else None,
+            "publish_to_drain_ms": round(at["drained"] - at["published"], 3)
+                if at["drained"] is not None and at["published"] is not None else None,
+            "worker_id": int(started.get("worker_id", -1)),
+        })
+    return rows
+
+
+def merge_seconds(intervals: list[tuple[float, float]], lo: float, hi: float) -> float:
+    clipped = sorted((max(lo, a), min(hi, b)) for a, b in intervals if b > lo and a < hi)
+    if not clipped:
+        return 0.0
+    total = 0.0
+    start, end = clipped[0]
+    for a, b in clipped[1:]:
+        if a <= end:
+            end = max(end, b)
+        else:
+            total += end - start
+            start, end = a, b
+    return total + end - start
+
+
+def analyze_trace(path: Path, phase: str, stratum: str) -> tuple[dict, list[dict], list[dict], dict]:
+    events = read_jsonl(path)
+    desired = next(e for e in events if e.get("type") == "desired_set")
+    settled = first_settled_sample(events)
+    first_admit = next(e for e in events if e.get("type") == "scheduler_admission"
+                       and e.get("desired_rebuilt") == 1)
+    start, finish = float(desired["t_s"]), float(settled["t_s"])
+    cfg = next(e for e in events if e.get("type") == "worker_backend_config")
+    requests = [e for e in events if e.get("type") == "request" and e.get("epoch") == 0
+                and e.get("level", 9) <= 2 and start <= e.get("t_s", -1) <= finish]
+    tiles = [e for e in events if e.get("type") == "tile" and e.get("level", 9) <= 2
+             and e.get("t_s", 1e99) <= finish]
+    by_ordinal = {e.get("request_ordinal"): e for e in tiles}
+    collision_by_key: dict[tuple[int, int, int, int], list[dict]] = defaultdict(list)
+    for e in events:
+        if e.get("type") == "collision_ready":
+            key = tuple(e.get("tile", [])) + (e.get("level", 0),)
+            collision_by_key[key].append(e)
+        elif e.get("type") == "collision_empty_classification":
+            key = tuple(e.get("key", [])) + (e.get("lod", 0),)
+            collision_by_key[key].append(e)
+    used_collision: dict[tuple[int, int, int, int], float] = defaultdict(float)
+    timeline = []
+    intervals_by_worker: dict[int, list[tuple[float, float]]] = defaultdict(list)
+    admitted_keys = set()
+    near_service = core_service = 0.0
+    for request in requests:
+        tile = by_ordinal.get(request.get("request_ordinal"))
+        if not tile:
+            continue
+        key = tuple(request["tile"]) + (request["level"],)
+        admitted_keys.add(key)
+        worker_start = tile.get("worker_start_t_s")
+        worker_end = tile.get("worker_end_t_s")
+        apply_t = tile.get("t_s")
+        collision = None
+        if apply_t is not None:
+            for candidate in collision_by_key.get(key, []):
+                candidate_t = float(candidate.get("t_s", -1))
+                if candidate_t >= float(apply_t) and candidate_t > used_collision[key]:
+                    collision = candidate
+                    used_collision[key] = candidate_t
+                    break
+        queue_at = worker_end
+        event_type = collision.get("type") if collision else (
+            "no_cook" if request["level"] > 0 else "pending_or_unobserved")
+        timeline.append({
+            "phase": phase, "stratum": stratum, "run": path.stem,
+            "x": key[0], "y": key[1], "z": key[2], "lod": key[3],
+            "request_ordinal": request.get("request_ordinal"),
+            "cohort_desired_t_s": round(start, 6),
+            "admitted_t_s": request.get("t_s"),
+            "worker_start_t_s": worker_start, "worker_end_t_s": worker_end,
+            "result_queued_approx_t_s": queue_at,
+            "applied_mesh_t_s": apply_t,
+            "collision_event": event_type,
+            "collision_ready_t_s": (collision.get("t_s") if collision
+                                     and collision.get("type") == "collision_ready" else None),
+            "collision_classified_t_s": (collision.get("t_s") if collision
+                                          and collision.get("type") == "collision_empty_classification" else None),
+            "worker_queue_s": tile.get("worker_queue_s"),
+            "worker_task_s": tile.get("worker_task_s"),
+            "result_queue_s": tile.get("result_queue_s"),
+            "core_generate_s": tile.get("core_generate_s"),
+            "mesh_copy_s": tile.get("mesh_copy_s"),
+            "rmc_stream_build_s": tile.get("rmc_stream_build_s"),
+            "apply_s": tile.get("apply_s"),
+            "worker_thread_id": tile.get("worker_thread_id"),
+            "empty": tile.get("empty"),
+        })
+        if worker_start is not None and worker_end is not None:
+            tid = int(tile.get("worker_thread_id", -1))
+            if tid >= 0:
+                intervals_by_worker[tid].append((float(worker_start), float(worker_end)))
+        near_service += float(tile.get("worker_task_s", 0.0))
+        core_service += float(tile.get("core_generate_s", 0.0))
+
+    sched_by_frame: dict[int, list[dict]] = defaultdict(list)
+    apply_by_frame: dict[int, list[dict]] = defaultdict(list)
+    queue_by_frame: dict[int, list[dict]] = defaultdict(list)
+    for e in events:
+        if not float(first_admit["t_s"]) <= e.get("t_s", -1) <= finish:
+            continue
+        if e.get("type") == "scheduler_admission":
+            sched_by_frame[int(e.get("frame", -1))].append(e)
+        elif e.get("type") == "apply_frame":
+            apply_by_frame[int(e.get("frame", -1))].append(e)
+        elif e.get("type") == "result_queue_frame":
+            queue_by_frame[int(e.get("frame", -1))].append(e)
+    frames = []
+    for frame in sorted(set(sched_by_frame) | set(apply_by_frame) | set(queue_by_frame)):
+        sched, applies, queues = sched_by_frame[frame], apply_by_frame[frame], queue_by_frame[frame]
+        submitted = [sum(e.get("submitted_by_lod", [0] * 9)[lod] for e in sched)
+                     for lod in range(9)]
+        apply_results = [sum(e.get("apply_results_by_lod", [0] * 9)[lod] for e in applies)
+                         for lod in range(9)]
+        meshes = [sum(e.get("meshes_by_lod", [0] * 9)[lod] for e in applies)
+                  for lod in range(9)]
+        t_values = [e.get("t_s") for e in sched + applies + queues if e.get("t_s") is not None]
+        frames.append({
+            "phase": phase, "stratum": stratum, "run": path.stem, "frame": frame,
+            "t_s_first_event": min(t_values) if t_values else None,
+            "scheduler_calls": len(sched),
+            "admitted_lod0_2": sum(submitted[:3]),
+            "admitted_by_lod": json.dumps(submitted, separators=(",", ":")),
+            "running_workers_avg": (round(statistics.mean(e.get("running_workers", 0)
+                                                           for e in sched), 3) if sched else None),
+            "apply_results_lod0_2": sum(apply_results[:3]),
+            "meshes_applied_lod0_2": sum(meshes[:3]),
+            "apply_ms": round(sum(e.get("apply_ms", 0.0) for e in applies), 4),
+            "frame_ms": max((e.get("frame_ms", 0.0) for e in applies), default=None),
+            "apply_budget_ms": max((e.get("apply_budget_ms", 0.0) for e in applies), default=None),
+            "queued_results_max": max((e.get("queued_results", 0) for e in queues), default=None),
+            "backlog_max": max((e.get("backlog_count", 0) for e in queues), default=None),
+        })
+
+    worker_rows = []
+    for thread_id, intervals in sorted(intervals_by_worker.items()):
+        busy = merge_seconds(intervals, float(first_admit["t_s"]), finish)
+        window = max(0.0, finish - float(first_admit["t_s"]))
+        worker_rows.append({
+            "phase": phase, "stratum": stratum, "run": path.stem,
+            "worker_thread_id": thread_id,
+            "near_task_busy_s": round(busy, 6),
+            "no_recorded_near_task_s": round(max(0.0, window - busy), 6),
+            "admission_to_settle_window_s": round(window, 6),
+            "near_task_busy_pct": round(100.0 * busy / window, 2) if window else 0.0,
+            "worker_task_count": sum(1 for e in tiles
+                                      if int(e.get("worker_thread_id", -1)) == thread_id),
+        })
+
+    summary = {
+        "phase": phase, "stratum": stratum, "run": path.stem,
+        "desired_t_s": start, "first_admission_t_s": float(first_admit["t_s"]),
+        "settled_sample_t_s": finish,
+        "admission_to_settle_s": finish - float(first_admit["t_s"]),
+        "desired_counts_lod0_2": settled.get("desired_by_lod", [0] * 9)[:3],
+        "settled_counts_lod0_2": settled.get("settled_by_lod", [0] * 9)[:3],
+        "unique_admitted_keys": len(admitted_keys), "tile_events": len(timeline),
+        "near_worker_s": near_service, "near_core_s": core_service,
+        "effective_workers": cfg.get("effective_tile_slots"),
+        "taskgraph_background_workers": cfg.get("taskgraph_background_threads"),
+        "ready_queue_depth_cvar": cfg.get("ready_queue_depth_cvar"),
+        "effective_ready_queue_depth": cfg.get("ready_queue_depth"),
+        "total_admission_cap": next(e.get("total_admission_cap") for e in events
+                                     if e.get("type") == "scheduler_admission"),
+        "result_backlog_cap": cfg.get("streaming_result_backlog_cap"),
+        "max_result_queue": max((r["queued_results_max"] or 0 for r in frames), default=0),
+        "max_result_backlog": max((r["backlog_max"] or 0 for r in frames), default=0),
+        "frame_count": len(frames),
+        "max_admit_per_engine_frame": max((r["admitted_lod0_2"] for r in frames), default=0),
+        "max_apply_meshes_per_engine_frame": max((r["meshes_applied_lod0_2"] for r in frames), default=0),
+        "max_apply_ms": max((r["apply_ms"] for r in frames), default=0),
+        "max_apply_budget_ms": max((r["apply_budget_ms"] or 0 for r in frames), default=0),
+        "p95_frame_ms": (sorted(r["frame_ms"] for r in frames if r["frame_ms"] is not None)
+                         [int(.95 * (len([r for r in frames if r["frame_ms"] is not None]) - 1))]
+                         if any(r["frame_ms"] is not None for r in frames) else None),
+        "max_frame_ms": max((r["frame_ms"] or 0 for r in frames), default=0),
+    }
+    return summary, timeline, frames, worker_rows
+
+
+def matched_worker_rows(trace: Path, phase: str, stratum: str,
+                        headless_runs: list[Path]) -> list[dict]:
+    events = read_jsonl(trace)
+    settled = first_settled_sample(events)
+    finish = float(settled["t_s"])
+    requests = [e for e in events if e.get("type") == "request" and e.get("epoch") == 0
+                and e.get("level", 9) <= 2 and e.get("t_s", 1e99) <= finish]
+    tile_events = [e for e in events if e.get("type") == "tile" and e.get("level", 9) <= 2
+                   and e.get("t_s", 1e99) <= finish]
+    tile_by_ordinal = {e.get("request_ordinal"): e for e in tile_events}
+    request_keys = {tuple(e["tile"]) + (e["level"],) for e in requests}
+    headless = [(path, headless_tiles(path)) for path in headless_runs]
+    reference_keys = set(headless[0][1])
+    common = reference_keys & request_keys
+    headless_s = [sum(row["worker_ms"] for key, row in tiles.items() if key in common) / 1000.0
+                  for _, tiles in headless]
+    matched = [tile_by_ordinal[e.get("request_ordinal")] for e in requests
+               if e.get("request_ordinal") in tile_by_ordinal
+               and tuple(e["tile"]) + (e["level"],) in common]
+    total_worker = sum(float(e.get("worker_task_s", 0.0)) for e in matched)
+    total_core = sum(float(e.get("core_generate_s", 0.0)) for e in matched)
+    total_copy = sum(float(e.get("mesh_copy_s", 0.0)) for e in matched)
+    total_rmc = sum(float(e.get("rmc_stream_build_s", 0.0)) for e in matched)
+    return [{
+        "phase": phase, "stratum": stratum, "run": trace.stem,
+        "headless_key_count": len(reference_keys), "common_key_count": len(common),
+        "unreal_only_key_count": len(request_keys - reference_keys),
+        "headless_only_key_count": len(reference_keys - request_keys),
+        "unreal_task_count_on_common_keys": len(matched),
+        "unreal_worker_seconds_on_common_keys": round(total_worker, 6),
+        "unreal_core_seconds_on_common_keys": round(total_core, 6),
+        "unreal_mesh_copy_seconds_on_common_keys": round(total_copy, 6),
+        "unreal_rmc_stream_build_seconds_on_common_keys": round(total_rmc, 6),
+        "unreal_worker_ms_per_common_key": round(total_worker * 1000.0 / len(common), 6)
+            if common else None,
+        "headless_worker_seconds_median_on_common_keys": round(statistics.median(headless_s), 6),
+        "headless_worker_ms_per_common_key": round(statistics.median(headless_s) * 1000.0 / len(common), 6)
+            if common else None,
+        "headless_ready_s_median": round(statistics.median(
+            float(re.search(r"cold_ready_s=([0-9.]+)", path.read_text(
+                encoding="utf-8", errors="replace")).group(1)) for path in headless_runs), 6),
+    }]
+
+
+def main() -> None:
+    OUT.mkdir(parents=True, exist_ok=True)
+    all_summaries, all_tiles, all_frames, all_workers = [], [], [], []
+    all_headless_tiles, all_matched_workers = [], []
+    headless_paths = {
+        "FI": sorted(HEADLESS.glob("fi-plain-w12-r*.log")),
+        "SW": sorted(HEADLESS.glob("sw-plain-w12-r*.log")),
+    }
+    for stratum, paths in headless_paths.items():
+        for path in paths:
+            all_headless_tiles.extend(headless_lifecycle(path))
+    for phase, directory, patterns in (
+        ("before", BEFORE, ("after-fi-on-*.trace.jsonl", "after-sw-on-*.trace.jsonl")),
+        ("after", AFTER, ("after-fi-rd64-bg1-boost1-*.trace.jsonl",
+                           "after-sw-rd64-bg1-boost1-*.trace.jsonl")),
+    ):
+        traces = sorted({p for pattern in patterns for p in directory.glob(pattern)})
+        for path in traces:
+            stratum = "FI" if "fi-" in path.name else "SW"
+            summary, tiles, frames, workers = analyze_trace(path, phase, stratum)
+            all_summaries.append(summary)
+            all_tiles.extend(tiles)
+            all_frames.extend(frames)
+            all_workers.extend(workers)
+            all_matched_workers.extend(matched_worker_rows(
+                path, phase, stratum, headless_paths[stratum]))
+            print("[L95Timeline] " + json.dumps(summary, sort_keys=True), flush=True)
+    write_csv(OUT / "unreal-tile-lifecycle.csv", all_tiles)
+    write_csv(OUT / "unreal-per-frame.csv", all_frames)
+    write_csv(OUT / "unreal-worker-occupancy.csv", all_workers)
+    write_csv(OUT / "headless-tile-lifecycle.csv", all_headless_tiles)
+    write_csv(OUT / "matched-worker-seconds.csv", all_matched_workers)
+    (OUT / "unreal-run-summary.json").write_text(
+        json.dumps(all_summaries, indent=2) + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()

Standalone/Research/WorldAssemblerM3/Linux95/REPORT.md

@@ -0,0 +1,41 @@
+# Linux L95: Unreal near-tile streaming
+
+## Timeline and cause
+
+The Unreal traces cover the initial LOD0–2 desired set. The lifecycle CSV records each tile's common desired-set time, admission (`request`), worker start/end, approximate result enqueue (worker end is recorded immediately before enqueue), mesh application, and collision-ready or empty classification. Per-frame admission/apply counts and per-worker near-task intervals are also exported under `Saved/Linux/l95/timelines/`.
+
+Per-key demand time is not emitted by Unreal, so the desired-set timestamp is a shared cohort origin. The `tile` event is after game-thread mesh submission to RMC, not GPU completion. `settled_by_lod` counts loaded or collision-not-required tiles; the first full sample is sampled at roughly 0.25 s intervals and does not mean every collision cook is finished. FI had 77 LOD0 collision-ready tiles by that sample; SW had 13–14 in normal runs. The 9-tile player support collision gate is measured separately.
+
+Admission dominated the critical path. Before, FI admissions spanned 2.45–2.51 s and SW 2.10–2.28 s. After, FI spanned 1.95–1.99 s and normal SW runs 1.21–1.27 s. The last result was applied 18–22 ms after admission in the after runs (1–20 ms before); the full-settled sample followed shortly after. The trace has at most one scheduler pass per engine frame. Nonzero admission frames fell from 109 to 84 for FI and from 97–101 to 46–49 for normal SW runs. LOD0, then LOD1, then LOD2 became resident, so near-first ordering was already working. Per-worker “idle” means no recorded LOD0–2 task; the thread may have done far-level or other engine work.
+
+## Worker service
+
+Headless runs imported the exact Unreal Core snapshots (world hash `1ecc162755ab83ac`, layout `0589300eb07dfe06`) and used 12 workers without `--profile-lod-work`. Service compares the exact shared LOD0–2 keys; Unreal has 22 additional SW keys, excluded from that comparison.
+
+| Stratum | Shared keys | Unreal before | Unreal after | Headless W12 |
+|---|---:|---:|---:|---:|
+| FI | 1,216 | 11.50 ms/tile | 11.24 ms/tile | 16.18 ms/tile |
+| SW | 1,408 | 1.00 ms/tile | 0.75 ms/tile | 1.13 ms/tile |
+
+These are summed worker-task intervals per shared key, not exclusive CPU cycles. They show no evidence that Unreal tile work is slower; this comparison does not isolate compiler or allocator differences between the host builds. Unreal had 8 effective TaskGraph tile workers versus headless's 12. Service-derived average occupied near slots rose from 5.2 to 6.5 of 8 for FI; for normal SW it was 0.6 to 0.8 of 8, versus 2.5 of 12 headless. This, plus tick-paced admission, explains the remaining wall-time gap better than per-tile adapter work. Unreal's worker time was nearly all Core generation; mesh copy plus RMC stream build totaled about 0.07 s across the FI cohort.
+
+The fix raises `StreamingReadyQueueDepth`'s default from 16 to 64. With 8 worker slots the existing clamp makes the effective depth 32 and total outstanding cap 40. The L94 before traces explicitly used depth 8/cap 16; the L95 after traces used 32/cap 40. The result backlog cap remains 256 and stayed below 102 in ordinary after runs (one SW outlier reached 250). Scheduler policy 3, dynamic near reserve, the 24-mesh/14 ms startup apply boost, and collision settings are unchanged. The movement portion exercised the walking case at 140 cm/s with policy 3 and the walkable gate ready; the fast-speed policy code is unchanged.
+
+## Before and after
+
+Times are first scheduler admission to first full LOD0–2 settled sample. Frame values are the trace's interval between apply-drain entries, so they include work outside the apply drain.
+
+| Arrival | Unreal before | Unreal after | `frame_ms` p95 before → after | max before → after |
+|---|---:|---:|---:|---:|
+| FI | 2.715 s (2.619–2.750, n=3) | 2.104 s (2.068–2.114, n=3) | 19.7–21.2 → 20.8–21.3 ms | 137–158 → 118–122 ms |
+| SW | 2.372 s (2.333–2.373, n=3) | 1.312 s (1.299–6.897, n=4) | 20.7–23.4 → 24.7–27.6 ms | 137–146 → 118–145 ms |
+
+Three SW after runs settled in 1.299–1.317 s; the fourth is a valid 6.897 s outlier. `apply_ms` stayed below the 14 ms budget in every run: 8.28 ms maximum before and 4.23 ms after. The SW tick-interval p95 is modestly higher after; it is not an isolated game-thread apply measurement.
+
+Fast tier passed in 296.8 s. Worker-count hash checks were stable, and no generation or passage-composer code changed. Linux builds completed before timing: headless Clang 19, FP contraction off, 4 jobs; Unreal Linux Development with `-NoUBA -MaxParallelActions=4`. Timed Unreal runs had no `--as`; headless runs used `prlimit --as=6442450944`; runs were pinned to CPUs 0–11. No builds ran during timing.
+
+## Files and failed runs
+
+Changed source: `Source/VoxelForge/Private/VoxelWorld.cpp` (queue default and opt-in snapshot export), `Standalone/Source/StreamBenchmarkHeadless.cpp` (exact snapshot input). Timeline outputs: `unreal-tile-lifecycle.csv`, `unreal-per-frame.csv`, `unreal-worker-occupancy.csv`, `headless-tile-lifecycle.csv`, and `matched-worker-seconds.csv` in `Saved/Linux/l95/timelines/`.
+
+No L95 timed run failed. The SW outlier repeatedly re-admitted three LOD0 keys and nearly filled the result backlog; its cause is undetermined, and it is included above. L94 SW repeat 3 did not reach arrival and was excluded from the before sample; repeats 1, 2, and 4 were used. Fast tier's two negative-control cases printed expected `FAIL` results, their rejection checks passed, and the tier finished PASS. Diagnostic snapshot-export and exploratory probe runs are excluded from timing samples.

Standalone/Research/WorldAssemblerM3/Linux95/RunL95UnrealArrival.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+"""Run one pinned null-RHI SW/FI arrival for Linux 95."""
+from __future__ import annotations
+
+import argparse
+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/l95"
+HOST_SAVED = HOST / "Saved/Linux/l95"
+EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd")
+OUT = ROOT / "Saved/Linux/l95/unreal-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
+
+
+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 run(stratum: str, repeat: int, phase: str, ready_depth: int,
+        background_cap: int, startup_boost: int,
+        snapshot_file: Path | None) -> dict:
+    label = (f"{phase}-{stratum.lower()}-rd{ready_depth}-bg{background_cap}"
+             f"-boost{startup_boost}-{repeat}")
+    trace = OUT / f"{label}.trace.jsonl"
+    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)
+    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}",
+        "-voxel.CoreColumnCacheDirectMapped=0", "-voxel.CoreMCGradientMemoization=1",
+        "-voxel.StreamingCollisionLookahead=1",
+        f"-voxel.StreamingStartupApplyBoost={startup_boost}",
+        "-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 snapshot_file is not None:
+        snapshot_file.parent.mkdir(parents=True, exist_ok=True)
+        command.append(f"-voxel.StartupTraceCoreSnapshotFile={snapshot_file}")
+    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 = read_events(trace)
+    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" and 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 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]
+    near_frames = [e for e in events if e.get("type") == "apply_frame" and first_admission
+                   and full and first_admission["t_s"] <= e.get("t_s", -1) <= full["t_s"]]
+    near_sched = [e for e in events if e.get("type") == "scheduler_admission" and first_admission
+                  and full and first_admission["t_s"] <= e.get("t_s", -1) <= full["t_s"]]
+    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]
+    max_req = max((e["t_s"] for e in near_requests), default=None)
+    row = {
+        "label": label, "stratum": stratum, "repeat": repeat, "phase": phase,
+        "ready_queue_depth": ready_depth, "background_worker_cap": background_cap,
+        "startup_apply_boost": startup_boost, "exit_code": result.returncode,
+        "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,
+        "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),
+        "last_admission_t_s": max_req,
+        "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_frames),
+        "burst_frame_ms_p95": round(sorted(e.get("frame_ms", 0.0) for e in near_frames)[
+            min(len(near_frames) - 1, int(.95 * (len(near_frames) - 1)))], 3)
+            if near_frames else None,
+        "burst_frame_ms_max": round(max((e.get("frame_ms", 0.0) for e in near_frames), default=0.0), 3),
+        "burst_max_meshes_per_frame": max((e.get("meshes_applied", 0) for e in near_frames), default=0),
+        "burst_max_apply_ms": round(max((e.get("apply_ms", 0.0) for e in near_frames), default=0.0), 3),
+        "burst_max_apply_budget_ms": max((e.get("apply_budget_ms", 0.0) for e in near_frames), default=0.0),
+        "burst_max_admit_per_frame": max((sum(e.get("submitted_by_lod", [0] * 9)[:3])
+                                            for e in near_sched), default=0),
+        "trace": str(trace.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")
+    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("[L95UnrealArrival] " + json.dumps(row, sort_keys=True), flush=True)
+    if result.returncode:
+        raise SystemExit(f"Unreal commandlet failed: {stdout}; exit={result.returncode}")
+    if not full or len(arrival) < 8:
+        raise SystemExit(f"Unreal arrival trace incomplete: {stdout}; exit={result.returncode}")
+    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("--startup-boost", type=int, choices=(0, 1), default=1)
+    parser.add_argument("--snapshot-file", type=Path)
+    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.startup_boost,
+        args.snapshot_file.resolve() if args.snapshot_file else None)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

Standalone/Source/StreamBenchmarkHeadless.cpp

@@ -20,6 +20,7 @@
 #include <limits>
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <tuple>
@@ -100,6 +101,7 @@ namespace
         std::filesystem::path MeshExportPath;
         std::filesystem::path WorldAssemblyManifestOutputPath;
         std::filesystem::path ResolvedLayoutOutputPath;
+        std::filesystem::path CoreSnapshotInputPath;
         bool ResolvedLayoutOnly=false;
         bool AuditResolvedLayoutBoundaries=false;
         bool AuditResolvedLayoutPassages=false;
@@ -1426,6 +1428,7 @@ namespace
                     "  --mesh-export=FILE (write the exact generated tile meshes for offline rendering)\n"
                     "  --capture-skip-walkability (mesh export only; defer the owner-wide passage gate)\n"
                     "  --world-assembler-manifest-output=FILE (write the canonical Core recipe and plan)\n"
+                    "  --core-snapshot=FILE (benchmark the exact resolved Core snapshot from Unreal)\n"
                     "  --profile-column-overlap --far-horizon-lod8-step=N --force-generate\n"
                     "  --seed=N --recipe=surface-world|floating-islands|caves|lattice\n"
                     "  --lattice-strate=N (opt one Maze or VerticalShafts strate into Core.Lattice)\n"
@@ -1478,6 +1481,9 @@ namespace
             else if(arg.starts_with("--world-assembler-manifest-output="))
                 options.WorldAssemblyManifestOutputPath=std::filesystem::path(std::string(
                     arg.substr(std::string_view("--world-assembler-manifest-output=").size())));
+            else if(arg.starts_with("--core-snapshot="))
+                options.CoreSnapshotInputPath=std::filesystem::path(std::string(
+                    arg.substr(std::string_view("--core-snapshot=").size())));
             else if(arg=="--resolved-layout-only")options.ResolvedLayoutOnly=true;
             else if(arg=="--audit-resolved-layout-boundaries")options.AuditResolvedLayoutBoundaries=true;
             else if(arg=="--audit-resolved-layout-passages")options.AuditResolvedLayoutPassages=true;
@@ -2249,10 +2255,51 @@ int main(int argc,char** argv)
     std::printf("[WorldAssemblerPlanTiming] total_ms=%.3f status=PASS\n",
         std::chrono::duration<double,std::milli>(Clock::now()-planStart).count());
 
-    auto snapshot=VoxelForgeCore::MakeStandaloneSnapshotFromParsedJson(ownerJsonRoot,seed,
+    VoxelForgeCore::ResolvedWorldSnapshot snapshot;
+    if (options.CoreSnapshotInputPath.empty())
+    {
+        snapshot = VoxelForgeCore::MakeStandaloneSnapshotFromParsedJson(ownerJsonRoot, seed,
             options.ClipRadius, options.MaxClipLevel, false, false, error, useExportedCorePlan);
         if (!snapshot)
-    {std::fprintf(stderr,"Shape interval build failed: %s\n",error.c_str());return 2;}
+        {
+            std::fprintf(stderr, "Shape interval build failed: %s\n", error.c_str());
+            return 2;
+        }
+    }
+    else
+    {
+        std::ifstream snapshotFile(options.CoreSnapshotInputPath, std::ios::binary);
+        if (!snapshotFile)
+        {
+            std::fprintf(stderr, "Core snapshot load failed: cannot open %s\n",
+                options.CoreSnapshotInputPath.string().c_str());
+            return 2;
+        }
+        const std::string snapshotJson{std::istreambuf_iterator<char>(snapshotFile),
+            std::istreambuf_iterator<char>{}};
+        if (!snapshotFile && !snapshotFile.eof())
+        {
+            std::fprintf(stderr, "Core snapshot load failed: read error for %s\n",
+                options.CoreSnapshotInputPath.string().c_str());
+            return 2;
+        }
+        std::optional<VoxelForgeCore::ResolvedWorldSnapshot> ImportedSnapshot =
+            VoxelForgeCore::ResolvedWorldSnapshot::FromJson(snapshotJson, error);
+        if (!ImportedSnapshot || ImportedSnapshot->worldAssembly() == nullptr
+            || ImportedSnapshot->resolvedLayout() == nullptr)
+        {
+            std::fprintf(stderr, "Core snapshot load failed: %s\n",
+                error.empty() ? "world assembly or resolved layout is missing" : error.c_str());
+            return 2;
+        }
+        snapshot = std::move(*ImportedSnapshot);
+        worldAssembly = *snapshot.worldAssembly();
+        resolvedLayout = *snapshot.resolvedLayout();
+        std::printf("[CoreSnapshotInput] path=%s world_assembly_hash=%016llx layout_identity=%s status=PASS\n",
+            options.CoreSnapshotInputPath.string().c_str(),
+            static_cast<unsigned long long>(worldAssembly.contentHash),
+            resolvedLayout.stableIdentity.c_str());
+    }
 
     FStandaloneCoreScheduler scheduler(static_cast<std::size_t>(options.ShapeHelperLimit));
     const FVector cameraStepVoxels=options.HasCameraStep