passages: contiguous BVH node ranges + child-bound reuse (L83); scaling and serial floor measured

Exact through LOD2 (all near tile hashes equal L82). W8 LOD0-2: SW 1.09->1.06, FI 2.47->2.34, TN 2.39->2.43 (noise), VS 1.20->1.18 s. Fast tier PASS. Scaling W4/W8/W12/W16 (LOD0-2, FI): 3.59 / 2.48 / 2.33 / 2.34 s - flattens past W8-W12. Estimated infinite-worker floor ~1.0 s FI/TN = shape compile (0.35-0.41 s, not overlapped: no shape-free tiles in these views) + slowest single LOD0 tile service (0.51-0.59 s). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
2026-10-28 00:24:49 +01:00
1 parent f572108dc4
commit a81b5b7d06
@@ -4055,6 +4055,7 @@ private:
        std::array<double,3> minimum{},maximum{};
        double maximumRadius=0.0;
        std::size_t minimumSegment=0,maximumSegment=0;
+        bool segmentRangeContiguous=false;
        std::vector<std::size_t> segments;
        int left=-1,right=-1;
    };
@@ -4873,6 +4874,8 @@ private:
        node.maximumSegment=*std::max_element(segments.begin(),segments.end());
        node.segments=segments;
        std::sort(node.segments.begin(),node.segments.end());
+        node.segmentRangeContiguous=node.segments.size()
+            ==node.maximumSegment-node.minimumSegment+1;
        const int nodeIndex=static_cast<int>(PassageNearestNodes[passageIndex].size());
        PassageNearestNodes[passageIndex].push_back(std::move(node));
        if(segments.size()>4)
@@ -4918,6 +4921,10 @@ private:
            bool hasCandidate=false;
            if(node.left<0)
            {
+            if(node.segmentRangeContiguous)
+                hasCandidate=static_cast<std::ptrdiff_t>(node.maximumSegment)>=first
+                    &&static_cast<std::ptrdiff_t>(node.minimumSegment)<end;
+            else
                for(const std::size_t segment:node.segments)
                    hasCandidate=hasCandidate
                        ||(static_cast<std::ptrdiff_t>(segment)>=first
@@ -5510,8 +5517,8 @@ private:
    template<typename FVisit>
    void VisitNearestPassageSpatialNode(std::size_t passageIndex,int nodeIndex,
        std::size_t first,std::size_t end,double qx,double qy,double qz,int priority,
-        const std::vector<std::uint8_t>* activeNodes,double& bestDistance,
-        FVisit& visit) const
+        const std::vector<std::uint8_t>* activeNodes,double knownLowerBound,
+        double& bestDistance,FVisit& visit) const
    {
        if(passageIndex>=PassageNearestNodes.size()||nodeIndex<0)return;
        const auto& nodes=PassageNearestNodes[passageIndex];
@@ -5528,8 +5535,7 @@ private:
                candidate=std::lower_bound(node.segments.begin(),node.segments.end(),first);
            if(candidate==node.segments.end()||*candidate>=end)return;
        }
-        const double lowerBound=PassageNearestNodeLowerBound(node,qx,qy,qz);
-        if(lowerBound>bestDistance)return;
+        if(knownLowerBound>bestDistance)return;
        if(node.left<0)
        {
            for(auto it=candidate;it!=node.segments.end();++it)
@@ -5554,9 +5560,11 @@ private:
        const int nearChild=leftBound<=rightBound?node.left:node.right;
        const int farChild=leftBound<=rightBound?node.right:node.left;
        VisitNearestPassageSpatialNode(passageIndex,nearChild,first,end,qx,qy,qz,
-            priority,activeNodes,bestDistance,visit);
+            priority,activeNodes,leftBound<=rightBound?leftBound:rightBound,
+            bestDistance,visit);
        VisitNearestPassageSpatialNode(passageIndex,farChild,first,end,qx,qy,qz,
-            priority,activeNodes,bestDistance,visit);
+            priority,activeNodes,leftBound<=rightBound?rightBound:leftBound,
+            bestDistance,visit);
    }

    template<typename FVisit>
@@ -5570,9 +5578,18 @@ private:
        const auto* tileQueries=ActivePassageTileQueries;
        const auto* activeNodes=tileQueries&&tileQueries->matches(*this)
            ?tileQueries->segmentRangeNodeMask(passageIndex,first,last+1,priority):nullptr;
+        if(activeNodes&&(activeNodes->empty()||!(*activeNodes)[0]))return;
+        if(!activeNodes)
+        {
+            const auto& root=PassageNearestNodes[passageIndex][0];
+            if(root.maximumSegment<static_cast<std::size_t>(first)
+                ||root.minimumSegment>=static_cast<std::size_t>(last+1))return;
+        }
+        const double rootLowerBound=PassageNearestNodeLowerBound(
+            PassageNearestNodes[passageIndex][0],qx,qy,qz);
        VisitNearestPassageSpatialNode(passageIndex,0,static_cast<std::size_t>(first),
            static_cast<std::size_t>(last+1),qx,qy,qz,priority,activeNodes,
-            bestDistance,callback);
+            rootLowerBound,bestDistance,callback);
    }

    bool PassagePotentialWithinNode(std::size_t passageIndex,int nodeIndex,
@@ -5581,13 +5598,18 @@ private:
    {
        if(passageIndex>=PassageNearestNodes.size()||nodeIndex<0)return false;
        const auto& nodes=PassageNearestNodes[passageIndex];
-        const auto& node=nodes[static_cast<std::size_t>(nodeIndex)];
+        const std::size_t nodeOffset=static_cast<std::size_t>(nodeIndex);
+        if(nodeOffset>=nodes.size())return false;
+        const auto& node=nodes[nodeOffset];
        if(node.maximumSegment<first||node.minimumSegment>=end)return false;
+        if(!node.segmentRangeContiguous)
+        {
            const bool allSegmentsInRange=first<=node.minimumSegment
                &&node.maximumSegment<end;
            const auto candidate=allSegmentsInRange?node.segments.begin()
                :std::lower_bound(node.segments.begin(),node.segments.end(),first);
            if(candidate==node.segments.end()||*candidate>=end)return false;
+        }
        if(PassageNearestNodeLowerBound(node,qx,qy,qz)>signedDistanceLimit)return false;
        if(node.left<0)return true;
        const double leftBound=PassageNearestNodeLowerBound(
@@ -5691,10 +5713,11 @@ private:
    {
        const auto possible=[&](std::ptrdiff_t first,std::ptrdiff_t last)
        {
-            return first<=last&&passageIndex<PassageNearestNodes.size()
-                &&!PassageNearestNodes[passageIndex].empty()
-                &&PassagePotentialWithinNode(passageIndex,0,static_cast<std::size_t>(first),
-                    static_cast<std::size_t>(last+1),qx,qy,qz,signedDistanceLimit);
+            if(first>last||passageIndex>=PassageNearestNodes.size()
+                ||PassageNearestNodes[passageIndex].empty())return false;
+            return PassagePotentialWithinNode(passageIndex,0,
+                static_cast<std::size_t>(first),static_cast<std::size_t>(last+1),
+                qx,qy,qz,signedDistanceLimit);
        };
        return possible(carveBegin,carveEnd)
            ||possible(0,std::min<std::ptrdiff_t>(8,lowerMouthIndex)-1)
@@ -5710,11 +5733,15 @@ private:
        const auto& nodes=PassageNearestNodes[passageIndex];
        const auto& node=nodes[static_cast<std::size_t>(nodeIndex)];
        if(node.maximumSegment<first||node.minimumSegment>=end)return false;
+        auto candidate=node.segments.begin();
+        if(!node.segmentRangeContiguous)
+        {
            const bool allSegmentsInRange=first<=node.minimumSegment
                &&node.maximumSegment<end;
-        auto candidate=allSegmentsInRange?node.segments.begin()
+            candidate=allSegmentsInRange?node.segments.begin()
                :std::lower_bound(node.segments.begin(),node.segments.end(),first);
            if(candidate==node.segments.end()||*candidate>=end)return false;
+        }
        const double xyReach=0.39*qPerMetre;
        if(qx+xyReach<node.minimum[0]||qx-xyReach>node.maximum[0]
            ||qy+xyReach<node.minimum[1]||qy-xyReach>node.maximum[1])return false;
@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""Symbolize L77 samples and group the passage work by source term."""
+from __future__ import annotations
+
+import collections
+import json
+import re
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+OUT = ROOT / "Saved/Linux/l83/census"
+SAMPLE = re.compile(
+    r"^\[L77Sample\] pc=(0x\w+) lod=(\d+) .*?phase=(\S+) .*?weight=(\d+)$", re.M)
+SYMBOLIZER = "llvm-symbolizer-19"
+BINARIES = {
+    "pre": ROOT / "Saved/Linux/l82/profile-bin/VoxelForgeCoreHeadless-clang19",
+    "post": ROOT / "Saved/Linux/l83/profile-bin/VoxelForgeCoreHeadless-l83-profile-clang19",
+}
+
+
+def source_term(sample_phase: str, symbol: dict) -> str:
+    function = symbol.get("FunctionName", "")
+    file_name = symbol.get("FileName", "").rsplit("/", 1)[-1]
+    line = int(symbol.get("Line", 0))
+    if any(name in function for name in (
+        "VF_FieldCachedValueNoise", "VF_FieldValueNoise", "VF_FieldSurfaceRelief",
+        "VF_FieldPassageWallRelief", "VF_FieldNoiseValue", "VF_FieldNoiseHash",
+        "VF_FieldMix", "VF_FieldLerp", "VF_FieldFade")):
+        return "remaining relief"
+    if file_name == "ShapeWorldInternal.h" and (
+        1459 <= line <= 1486 or 1807 <= line <= 1843):
+        return "smooth joins / Catmull-Rom"
+    if file_name == "ShapeWorldInternal.h" and (
+        1417 <= line <= 1426 or 1591 <= line <= 1778):
+        return "mouth shaping"
+    if any(name in function for name in (
+        "PassageColumnSignedDistance", "PassageBridgeSignedDistance",
+        "PassagePlayerCapsuleBlocksSupport", "PassageOtherCapsuleBlocksSupport",
+        "PassageSupportMayBeCarvedByLaterBore")):
+        return "support restoration / columns"
+    if file_name == "ShapeWorldInternal.h" and (
+        1275 <= line <= 1337 or 1781 <= line <= 2200):
+        return "support restoration / columns"
+    if sample_phase == "support":
+        return "support restoration / columns"
+    if any(name in function for name in (
+        "PassagePotentialWithinNode", "VisitNearestPassageSpatialNode",
+        "PassageNearestNodeLowerBound", "PassageMayAffectDensity",
+        "BuildPassageSegmentRangeNodeMask", "segmentRangeNodeMask")):
+        return "carve distance evaluation"
+    if file_name == "ShapeWorldInternal.h" and (
+        1049 <= line <= 1179 or 1344 <= line <= 1538):
+        return "carve distance evaluation"
+    if line == 0 and "ApplyResolvedStratePassageCarves" in function:
+        return "passage body (line not resolved)"
+    return "other passage work"
+
+
+def analyze(label: str) -> dict:
+    samples_by_pc: collections.Counter[tuple[str, str, str]] = collections.Counter()
+    phase_ticks: collections.Counter[tuple[str, str]] = collections.Counter()
+    files = sorted(OUT.glob(f"{label}-fi-r*.log")) + sorted(OUT.glob(f"{label}-tn-r*.log"))
+    for path in files:
+        world = "FI" if "-fi-" in path.name else "TN"
+        output = path.read_text(encoding="utf-8", errors="replace")
+        for pc, lod, phase, weight in SAMPLE.findall(output):
+            if int(lod) > 2:
+                continue
+            ticks = int(weight)
+            phase_ticks[(world, phase)] += ticks
+            samples_by_pc[(world, pc, phase)] += ticks
+    pcs = sorted({pc for _world, pc, _phase in samples_by_pc})
+    command = [str(SYMBOLIZER), f"--obj={BINARIES[label]}",
+               "--output-style=JSON", "--no-inlines"]
+    result = subprocess.run(command, input="\n".join(pcs) + "\n", text=True,
+                            capture_output=True, check=True)
+    symbols = {}
+    for line in result.stdout.splitlines():
+        item = json.loads(line)
+        symbols[item["Address"]] = item.get("Symbol", [{}])[0]
+
+    worlds = {}
+    for world in ("FI", "TN"):
+        all_ticks = sum(value for (w, _phase), value in phase_ticks.items() if w == world)
+        passage_ticks = phase_ticks[(world, "passages")]
+        support_phase_ticks = phase_ticks[(world, "support")]
+        terms: collections.Counter[str] = collections.Counter()
+        top_symbols: collections.Counter[str] = collections.Counter()
+        for (sample_world, pc, phase), ticks in samples_by_pc.items():
+            if sample_world != world or phase != "passages":
+                continue
+            symbol = symbols.get(pc, {})
+            terms[source_term(phase, symbol)] += ticks
+            function = symbol.get("FunctionName", "<unresolved>")
+            top_symbols[function] += ticks
+        # Support is a separately tagged L77 phase; classify it here with relief kept
+        # separate by symbol, to avoid folding relief calls into support CPU.
+        for (sample_world, pc, phase), ticks in samples_by_pc.items():
+            if sample_world != world or phase != "support":
+                continue
+            symbol = symbols.get(pc, {})
+            term = source_term(phase, symbol)
+            if term == "remaining relief":
+                terms[term] += ticks
+            else:
+                terms["support restoration / columns"] += ticks
+        passage_plus_support_ticks = passage_ticks + support_phase_ticks
+        worlds[world] = {
+            "near_lod_total_ticks": all_ticks,
+            "passage_phase_ticks": passage_ticks,
+            "passage_phase_share_pct": 100.0 * passage_ticks / all_ticks,
+            "support_phase_ticks": support_phase_ticks,
+            "support_within_passage_plus_support_pct":
+                100.0 * support_phase_ticks / passage_plus_support_ticks,
+            "term_ticks": dict(terms),
+            "terms_pct_of_passage_and_support_phases": {
+                term: 100.0 * value / passage_plus_support_ticks
+                for term, value in terms.items()},
+            "top_symbols": [
+                {"symbol": symbol, "ticks": ticks,
+                 "pct_of_passage_phase": 100.0 * ticks / passage_ticks}
+                for symbol, ticks in top_symbols.most_common(14)],
+        }
+    return {"label": label, "worlds": worlds}
+
+
+def main() -> int:
+    result = {label: analyze(label) for label in ("pre", "post")}
+    (OUT / "analysis.json").write_text(json.dumps(result, indent=2) + "\n",
+                                        encoding="utf-8")
+    for label in ("pre", "post"):
+        for world, row in result[label]["worlds"].items():
+            print(json.dumps({"label": label, "world": world,
+                              "passage_share_pct": row["passage_phase_share_pct"],
+                              "support_phase_ticks": row["support_phase_ticks"],
+                              "terms_pct": row["terms_pct_of_passage_and_support_phases"],
+                              "top_symbols": row["top_symbols"][:8]}, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,75 @@
+# Linux L83 — passage CPU and worker scaling
+
+Base was L82 (`f572108`); the retained code is the exact L83 candidate. Logs and machine-readable results are under `Saved/Linux/l83/`. Every benchmark/headless process used CPUs 0–11; headless runs had a 6 GiB address-space cap and a 1,740 s timeout. This Intel i9-12900H host exposes six P cores / twelve P-core threads on that mask, so W16 is oversubscription. The Ryzen 5800X3D target was not measured directly.
+
+## L77 near-LOD census
+
+Three L77 samples per world and version. Passage phase is the share of all near-LOD0–2 samples. Term shares use passage plus support-phase samples; these are PC/source-line attribution estimates, not subroutine wall-clock timers. `unresolved` is code without a resolved source-line term.
+
+| Term share | FI before → after | TN before → after |
+|---|---:|---:|
+| Carve distance evaluation | 45.1% → 41.5% | 43.5% → 39.2% |
+| Support restoration / columns | 16.6% → 14.9% | 16.2% → 13.5% |
+| Mouth shaping | 4.3% → 4.2% | 4.0% → 4.5% |
+| Smooth joins / Catmull–Rom | 0.7% → 0.8% | 1.2% → 1.3% |
+| Remaining relief | 18.9% → 21.3% | 19.7% → 23.7% |
+| Other passage work | 9.3% → 11.5% | 10.2% → 11.5% |
+| Unresolved passage body | 5.1% → 5.8% | 5.1% → 6.3% |
+
+Passage phase itself was FI 58.76% → 58.75%, TN 58.51% → 59.11% of near-LOD samples. Top exclusive symbols (% of passage-phase samples):
+
+| World/version | Top symbols |
+|---|---|
+| FI before | `VisitNearestPassageSpatialNode` 24.3%; `ApplyResolvedStratePassageCarves` 21.4%; `VF_FieldCachedValueNoise` 18.7%; `PassagePotentialWithinNode` 17.5%; carve lambda 10.2% |
+| FI after | `ApplyResolvedStratePassageCarves` 21.3%; `VF_FieldCachedValueNoise` 20.8%; `VisitNearestPassageSpatialNode` 20.3%; `PassagePotentialWithinNode` 12.7%; carve lambda 11.8% |
+| TN before | `VisitNearestPassageSpatialNode` 25.9%; `ApplyResolvedStratePassageCarves` 21.3%; `VF_FieldCachedValueNoise` 19.2%; `PassagePotentialWithinNode` 14.2%; carve lambda 12.3% |
+| TN after | `VF_FieldCachedValueNoise` 22.8%; `ApplyResolvedStratePassageCarves` 21.8%; `VisitNearestPassageSpatialNode` 20.8%; carve lambda 13.1%; `PassagePotentialWithinNode` 10.5% |
+
+## Exact cut and W8 readiness
+
+Marked BVH nodes whose segment indices form a contiguous range, then skipped repeated `lower_bound`/membership work for those nodes. The nearest-node traversal now passes each already-computed child distance bound into recursion instead of calculating that bound again. A mask-only reachability trial was exact but showed mixed timing, so it was discarded.
+
+L80 protocol: one warmup, then three interleaved L82/L83 pairs per stratum at W8. Cells show L82 → L83 median seconds; parentheses show the median of paired deltas.
+
+| Strate | LOD0–2 ready | Whole view |
+|---|---:|---:|
+| SW | 1.088 → 1.058 (−0.021) | 2.323 → 2.249 (−0.074) |
+| FI | 2.474 → 2.341 (−0.133) | 3.756 → 3.591 (−0.165) |
+| TN | 2.386 → 2.431 (+0.058) | 3.542 → 3.534 (−0.008) |
+| VS | 1.197 → 1.181 (−0.043) | 2.233 → 2.171 (−0.109) |
+
+The cut is exact through LOD2: every near tile hash and the aggregate hash matched L82, with zero mismatches. Whole hashes (density / mesh / payload / topology):
+
+| Strate | L82 = L83 whole hash |
+|---|---|
+| SW (1,408 tiles) | `4889b882f274930e / cafd187b455569f1 / 09e265430de5d9c8 / 498dc21d1fb4100e` |
+| FI (1,216 tiles) | `3175bceb00bfc76c / d1dabfba70a95d8a / b23900f97cfbb2cf / 3967a02b7c1ae1c8` |
+| TN (1,344 tiles) | `6b3e9f38209c1d8f / efa0d7b993928cbc / ede5b7bf3a771552 / 798dc428930aa2e2` |
+| VS (1,344 tiles) | `e0c51b07a9a2016c / 99f2772314f721c5 / 922b324240786f79 / 6f8a836e0a17895c` |
+
+## Scaling
+
+Final L83, L80 protocol: one warmup and three interleaved measured runs for each W4/W8/W12/W16 setting and stratum. Entries are median `LOD0–2 / whole-view` seconds; individual runs are in `Saved/Linux/l83/scaling/summary.json`.
+
+| Strate | W4 | W8 | W12 | W16 |
+|---|---:|---:|---:|---:|
+| SW | 1.464 / 3.493 | 1.455 / 3.023 | 1.210 / 2.427 | 1.087 / 2.309 |
+| FI | 3.592 / 5.341 | 2.478 / 3.791 | 2.328 / 3.566 | 2.342 / 3.608 |
+| TN | 3.625 / 5.398 | 2.475 / 3.631 | 2.327 / 3.384 | 2.295 / 3.368 |
+| VS | 1.540 / 3.160 | 1.205 / 2.242 | 1.274 / 2.254 | 1.158 / 2.218 |
+
+FI/TN scale well from W4 to W8 and gain about 0.15 s near-ready at W12; W16 is essentially flat for FI and slightly better for TN. Whole-view results mostly flatten by W12. W16 does not consistently help across all strata. Three-run variance is material in some cells: SW W4 near-ready ranged 1.400–2.322 s (shape compile 0.413–1.027 s), and FI W12 near-ready ranged 2.286–2.634 s (shape 0.344–0.594 s).
+
+## Serial floor
+
+At W16, the FI/TN timeline had `shape_free_desired=0` and `near_ring=0/27`. Shape compilation ran asynchronously, but there were no tiles to generate during it, so it is on the critical path and did not overlap tile generation in these views. Shape-free tiles could overlap compilation in views that actually contain them.
+
+An instrumented W16 pass estimates the infinite-worker floor as the maximum, over near tiles, of enqueue time plus that tile's measured service time: FI 1.014 s (last enqueue 0.430 s; maximum LOD0 service 0.587 s), TN 0.974 s (0.470 s; 0.506 s). This is an estimate from W16 service times, not a measured W∞ run. Shape compile in these passes was 0.347 s FI and 0.408 s TN; queue wait reached about 1.8 s, so the actual W16 runs still have substantial parallel backlog beyond that floor.
+
+## Verification and failed runs
+
+- Fast tier passed in 299.8 s; source stamp `08af4e98c86a69477cc79e21463d8c93bcd11ebe0f077d7c11519031d9141fa9`.
+- No Unreal build was needed: no Unreal `Source/` files changed.
+- The first L77 parser attempt failed on a census regex unpack error; the parser was corrected and all six before/after FI/TN captures completed.
+- The first fast-tier invocation stopped immediately because its output directory had no verification binary. After building that binary, the full tier passed.
+- No timed-out benchmark or verification run. No commit, push, or stash.
@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""Run three L77 sampler censuses per passage-heavy stratum."""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+OUT = ROOT / "Saved/Linux/l83/census"
+DATA = ROOT / "Standalone/data/owner_world.json"
+POST = len(sys.argv) > 1 and sys.argv[1] == "--post"
+BINARY = (ROOT / "Saved/Linux/l83/profile-bin/VoxelForgeCoreHeadless-l83-profile-clang19"
+          if POST else ROOT / "Saved/Linux/l82/profile-bin/VoxelForgeCoreHeadless-clang19")
+LABEL = "post" if POST else "pre"
+CAMERAS = {
+    "FI": (25.8369140625, 3.19921875, -737.0),
+    "TN": (19.166015625, -14.2001953125, -753.3740234375),
+}
+SAMPLE = re.compile(r"^\[L77Sample\] pc=(0x\w+) lod=(\d+) .*?phase=(\S+) .*?weight=(\d+)$", re.M)
+SAMPLER = re.compile(r"^\[L77Sampler\] action=stop .*?samples=(\d+).*?status=PASS$", re.M)
+
+
+def run(world: str, rep: int) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    log = OUT / f"{LABEL}-{world.lower()}-r{rep}.log"
+    camera = ",".join(f"{value:.12f}" for value in CAMERAS[world]) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "env", "VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS=0", "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", "--linux77-profile",
+    ]
+    start = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    output = log.read_text(encoding="utf-8", errors="replace")
+    samples = [(int(lod), phase, int(weight))
+               for _pc, lod, phase, weight in SAMPLE.findall(output)]
+    sampler = SAMPLER.search(output)
+    if result.returncode or not samples or sampler is None:
+        raise RuntimeError(f"census failed: exit={result.returncode} {log}")
+    near = [row for row in samples if row[0] <= 2]
+    passage = sum(weight for _, phase, weight in near if phase == "passages")
+    total = sum(weight for _, _, weight in near)
+    row = {
+        "world": world, "replicate": rep, "exit_code": result.returncode,
+        "wall_s": round(time.monotonic() - start, 3),
+        "sampler_samples": int(sampler.group(1)), "near_lod_samples": len(near),
+        "near_lod_weighted_ticks": total, "near_lod_passage_weighted_ticks": passage,
+        "near_lod_passage_share_pct": 100.0 * passage / total,
+        "log": str(log.relative_to(ROOT)),
+    }
+    print("[L83CensusRun] " + json.dumps(row, sort_keys=True), flush=True)
+    return row
+
+
+def main() -> int:
+    if not BINARY.is_file() or not DATA.is_file():
+        raise SystemExit(f"missing input: {BINARY} or {DATA}")
+    rows = [run(world, rep) for rep in (1, 2, 3) for world in CAMERAS]
+    by_world = {}
+    for world in CAMERAS:
+        values = [row["near_lod_passage_share_pct"] for row in rows if row["world"] == world]
+        by_world[world] = {"runs_pct": values, "mean_pct": sum(values) / len(values)}
+    (OUT / f"{LABEL}-summary.json").write_text(json.dumps(
+        {"runs": rows, "near_lod_passage_share": by_world}, indent=2) + "\n",
+        encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Three interleaved L80-protocol W8 readiness pairs for L82 vs final L83."""
+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/l83/readiness"
+DATA = ROOT / "Standalone/data/owner_world.json"
+BINARIES = {
+    "L82": ROOT / "Saved/Linux/l82/bin/VoxelForgeCoreHeadless-clang19",
+    "L83": ROOT / "Saved/Linux/l83/bin/VoxelForgeCoreHeadless-l83-clang19",
+}
+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),
+}
+EXPECTED_NEAR = {0: 1408, 1: 1216, 2: 1344, 3: 1344}
+EXPECTED_WHOLE = {0: 2850, 1: 2249, 2: 2233, 3: 2329}
+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+) "
+                  r"cold_ready_s=([0-9.]+)$", re.M)
+SHAPE_FREE = re.compile(r"^\[ShapeReadiness\] shape_free_desired=(\d+)/(\d+) "
+                        r"near_ring=(\d+)/27", re.M)
+STAMP = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=(\S+)$", re.M)
+EXPECTED_STAMPS = {
+    "L82": "99fcbd2fd46e0bb3e250f3ad67e98ade0545e419193f987550356015714c6f1b",
+    "L83": "08af4e98c86a69477cc79e21463d8c93bcd11ebe0f077d7c11519031d9141fa9",
+}
+
+
+def run(strate: int, version: str, phase: str, rep: int) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    log = OUT / f"s{strate}-{version.lower()}-{phase}{rep}.log"
+    camera = ",".join(f"{v:.12f}" for v in CAMERAS[strate]) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "env", "VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS=0", "timeout", "1740s",
+        str(BINARIES[version]), 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", "--stream-scheduler-policy=3",
+    ]
+    start = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    output = log.read_text(encoding="utf-8", errors="replace")
+    pipeline_match, near_match = PIPELINE.search(output), NEAR.search(output)
+    shape_free_match, stamp_match = SHAPE_FREE.search(output), STAMP.search(output)
+    if result.returncode or pipeline_match is None or near_match is None:
+        raise RuntimeError(f"readiness run failed: exit={result.returncode} {log}")
+    if stamp_match is None or stamp_match.group(1) != EXPECTED_STAMPS[version]:
+        raise RuntimeError(f"source stamp mismatch in {log}")
+    pipeline = dict(FIELDS.findall(pipeline_match.group(1)))
+    near_done, near_total = int(near_match.group(1)), int(near_match.group(2))
+    whole_done, whole_total = (int(v) for v in pipeline["whole_tiles"].split("/"))
+    if near_done != near_total or near_total != EXPECTED_NEAR[strate]:
+        raise RuntimeError(f"near readiness incomplete in {log}: {near_done}/{near_total}")
+    if whole_done != whole_total or whole_total != EXPECTED_WHOLE[strate]:
+        raise RuntimeError(f"whole readiness incomplete in {log}: {whole_done}/{whole_total}")
+    row = {
+        "strate": strate, "version": version, "phase": phase, "replicate": rep,
+        "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")),
+        "shape_s": float(pipeline["shape"].removesuffix("s")),
+        "first_tile_ready_s": float(pipeline["first_tile_ready"].removesuffix("s")),
+        "neighborhood_ready_s": float(pipeline["neighborhood_ready"].removesuffix("s")),
+        "near_tiles": near_total, "whole_tiles": whole_total,
+        "shape_free_desired": int(shape_free_match.group(1)) if shape_free_match else None,
+        "shape_free_near_ring": int(shape_free_match.group(3)) if shape_free_match else None,
+        "source_stamp": stamp_match.group(1), "log": str(log.relative_to(ROOT)),
+    }
+    print("[L83ReadinessRun] " + json.dumps(row, sort_keys=True), flush=True)
+    return row
+
+
+def main() -> int:
+    for path in (*BINARIES.values(), DATA):
+        if not path.is_file():
+            raise SystemExit(f"missing input: {path}")
+    for strate in range(4):
+        for version in ("L82", "L83"):
+            run(strate, version, "warmup", 0)
+    rows = []
+    orders = (("L82", "L83"), ("L83", "L82"), ("L82", "L83"))
+    for strate in range(4):
+        for rep, order in enumerate(orders, 1):
+            for version in order:
+                rows.append(run(strate, version, "measured", rep))
+                (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]
+        item = {"strate": strate, "measured_pairs": 3}
+        for version in ("L82", "L83"):
+            subset = [row for row in group if row["version"] == version]
+            for metric in ("lod02_ready_s", "whole_view_ready_s", "shape_s"):
+                values = [row[metric] for row in subset]
+                item[f"{version}_{metric}"] = values
+                item[f"{version}_{metric}_median"] = statistics.median(values)
+        for metric in ("lod02_ready_s", "whole_view_ready_s"):
+            by_pair = {rep: {row["version"]: row[metric] for row in group
+                             if row["phase"] == "measured" and row["replicate"] == rep}
+                       for rep in (1, 2, 3)}
+            deltas = [by_pair[rep]["L83"] - by_pair[rep]["L82"] for rep in (1, 2, 3)]
+            item[f"paired_{metric}_delta_s"] = deltas
+            item[f"paired_{metric}_delta_median_s"] = statistics.median(deltas)
+        summary.append(item)
+    (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())
@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Compare all LOD0-2 near tile hashes per stratum against L82."""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+OUT = ROOT / "Saved/Linux/l83/hash-check"
+DATA = ROOT / "Standalone/data/owner_world.json"
+BINARIES = {
+    "L82": ROOT / "Saved/Linux/l82/bin/VoxelForgeCoreHeadless-clang19",
+    "L83": ROOT / "Saved/Linux/l83/bin/VoxelForgeCoreHeadless-l83-clang19",
+}
+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),
+}
+EXPECTED_NEAR = {0: 1408, 1: 1216, 2: 1344, 3: 1344}
+TILE_HASH = re.compile(
+    r"^\[PipelineTileHash\] lod=(-?\d+) x=(-?\d+) y=(-?\d+) z=(-?\d+) "
+    r"density=(0x[0-9a-f]+) mesh=(0x[0-9a-f]+) payload=(0x[0-9a-f]+) "
+    r"topology=(0x[0-9a-f]+)$", re.M)
+WHOLE_HASH = re.compile(
+    r"^\[PipelineWholeViewHash\] tiles=(\d+) density=(0x[0-9a-f]+) "
+    r"mesh=(0x[0-9a-f]+) payload=(0x[0-9a-f]+) topology=(0x[0-9a-f]+)$", re.M)
+
+
+def run(strate: int, version: str) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    log = OUT / f"{version.lower()}-s{strate}.log"
+    camera = ",".join(f"{v:.12f}" for v in CAMERAS[strate]) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "env", "VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS=0", "timeout", "1740s",
+        str(BINARIES[version]), str(DATA), "--world-assembler-prototype",
+        "--stream-benchmark", "--stream-pipeline", "--workers=8", "--clip-radius=3",
+        "--max-clip-level=8", "--max-level=2", f"--camera={camera}",
+        "--stream-hash-tiles",
+    ]
+    start = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    output = log.read_text(encoding="utf-8", errors="replace")
+    hashes = {
+        (int(m.group(1)), int(m.group(2)), int(m.group(3)), int(m.group(4))): m.group(5, 6, 7, 8)
+        for m in TILE_HASH.finditer(output)
+    }
+    whole = WHOLE_HASH.search(output)
+    if result.returncode or whole is None or len(hashes) != EXPECTED_NEAR[strate]:
+        raise RuntimeError(f"hash run failed or incomplete: exit={result.returncode} {log}")
+    return {
+        "version": version, "strate": strate, "exit_code": result.returncode,
+        "wall_s": round(time.monotonic() - start, 3), "tile_hash_count": len(hashes),
+        "whole_hash": whole.group(1, 2, 3, 4, 5), "hashes": hashes,
+        "log": str(log.relative_to(ROOT)),
+    }
+
+
+def main() -> int:
+    for path in (*BINARIES.values(), DATA):
+        if not path.is_file():
+            raise SystemExit(f"missing input: {path}")
+    comparisons, runs = [], []
+    for strate in range(4):
+        pair = {version: run(strate, version) for version in ("L82", "L83")}
+        a, b = pair["L82"], pair["L83"]
+        mismatch_keys = sorted(key for key in set(a["hashes"]) | set(b["hashes"])
+                               if a["hashes"].get(key) != b["hashes"].get(key))
+        comparison = {
+            "strate": strate, "tiles": len(a["hashes"]),
+            "tile_hashes_identical": not mismatch_keys,
+            "whole_hash_identical": a["whole_hash"] == b["whole_hash"],
+            "mismatch_count": len(mismatch_keys), "first_mismatches": mismatch_keys[:8],
+            "l82_whole_hash": a["whole_hash"], "l83_whole_hash": b["whole_hash"],
+        }
+        comparisons.append(comparison)
+        runs.extend({k: v for k, v in row.items() if k != "hashes"} for row in pair.values())
+        print("[L83HashComparison] " + json.dumps(comparison, sort_keys=True), flush=True)
+    (OUT / "summary.json").write_text(
+        json.dumps({"comparisons": comparisons, "runs": runs}, indent=2) + "\n",
+        encoding="utf-8")
+    return 0 if all(row["tile_hashes_identical"] and row["whole_hash_identical"]
+                    for row in comparisons) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""L80-protocol cold readiness scaling at 4, 8, 12, and 16 workers."""
+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/l83/scaling"
+DATA = ROOT / "Standalone/data/owner_world.json"
+BINARY = ROOT / "Saved/Linux/l83/bin/VoxelForgeCoreHeadless-l83-clang19"
+WORKERS = (4, 8, 12, 16)
+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),
+}
+EXPECTED_NEAR = {0: 1408, 1: 1216, 2: 1344, 3: 1344}
+EXPECTED_WHOLE = {0: 2850, 1: 2249, 2: 2233, 3: 2329}
+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+) "
+                  r"cold_ready_s=([0-9.]+)$", re.M)
+SHAPE_FREE = re.compile(r"^\[ShapeReadiness\] shape_free_desired=(\d+)/(\d+) "
+                        r"near_ring=(\d+)/27", re.M)
+STAMP = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=(\S+)$", re.M)
+SOURCE_STAMP = "08af4e98c86a69477cc79e21463d8c93bcd11ebe0f077d7c11519031d9141fa9"
+
+
+def run(strate: int, workers: int, phase: str, rep: int) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    log = OUT / f"s{strate}-w{workers}-{phase}{rep}.log"
+    camera = ",".join(f"{v:.12f}" for v in CAMERAS[strate]) + ",0,0"
+    command = [
+        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
+        "env", "VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS=0", "timeout", "1740s",
+        str(BINARY), str(DATA), "--world-assembler-prototype", "--stream-benchmark",
+        "--stream-pipeline", f"--workers={workers}", "--clip-radius=3",
+        "--max-clip-level=8", "--max-level=8", f"--camera={camera}",
+        "--profile-lod-work", "--stream-scheduler-policy=3",
+    ]
+    start = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    output = log.read_text(encoding="utf-8", errors="replace")
+    pipeline_match, near_match = PIPELINE.search(output), NEAR.search(output)
+    shape_free_match, stamp_match = SHAPE_FREE.search(output), STAMP.search(output)
+    if result.returncode or pipeline_match is None or near_match is None:
+        raise RuntimeError(f"readiness run failed: exit={result.returncode} {log}")
+    if stamp_match is None or stamp_match.group(1) != SOURCE_STAMP:
+        raise RuntimeError(f"source stamp mismatch in {log}: {stamp_match.group(1) if stamp_match else 'missing'}")
+    pipeline = dict(FIELDS.findall(pipeline_match.group(1)))
+    near_done, near_total = int(near_match.group(1)), int(near_match.group(2))
+    whole_done, whole_total = (int(v) for v in pipeline["whole_tiles"].split("/"))
+    if near_done != near_total or near_total != EXPECTED_NEAR[strate]:
+        raise RuntimeError(f"near readiness incomplete in {log}: {near_done}/{near_total}")
+    if whole_done != whole_total or whole_total != EXPECTED_WHOLE[strate]:
+        raise RuntimeError(f"whole readiness incomplete in {log}: {whole_done}/{whole_total}")
+    row = {
+        "strate": strate, "workers": workers, "phase": phase, "replicate": rep,
+        "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")),
+        "shape_s": float(pipeline["shape"].removesuffix("s")),
+        "first_tile_ready_s": float(pipeline["first_tile_ready"].removesuffix("s")),
+        "neighborhood_ready_s": float(pipeline["neighborhood_ready"].removesuffix("s")),
+        "full_detail_ready_s": float(pipeline["full_detail_ready"].removesuffix("s")),
+        "near_tiles": near_total, "whole_tiles": whole_total,
+        "shape_free_desired": int(shape_free_match.group(1)) if shape_free_match else None,
+        "shape_free_near_ring": int(shape_free_match.group(3)) if shape_free_match else None,
+        "log": str(log.relative_to(ROOT)),
+    }
+    print("[L83ScalingRun] " + json.dumps(row, sort_keys=True), flush=True)
+    return row
+
+
+def main() -> int:
+    if not BINARY.is_file() or not DATA.is_file():
+        raise SystemExit(f"missing input: {BINARY} or {DATA}")
+    for strate in range(4):
+        for workers in WORKERS:
+            run(strate, workers, "warmup", 0)
+    rows = []
+    orders = (
+        WORKERS,
+        (16, 12, 8, 4),
+        (8, 16, 4, 12),
+    )
+    for strate in range(4):
+        for rep, order in enumerate(orders, 1):
+            for workers in order:
+                rows.append(run(strate, workers, "measured", rep))
+                (OUT / "runs-partial.json").write_text(
+                    json.dumps(rows, indent=2) + "\n", encoding="utf-8")
+    summary = []
+    for strate in range(4):
+        for workers in WORKERS:
+            subset = [row for row in rows
+                      if row["strate"] == strate and row["workers"] == workers]
+            item = {"strate": strate, "workers": workers, "measured_runs": len(subset)}
+            for metric in ("lod02_ready_s", "whole_view_ready_s", "shape_s",
+                           "first_tile_ready_s", "neighborhood_ready_s", "full_detail_ready_s"):
+                values = [row[metric] for row in subset]
+                item[metric] = values
+                item[f"{metric}_median"] = statistics.median(values)
+                item[f"{metric}_min"] = min(values)
+                item[f"{metric}_max"] = max(values)
+            summary.append(item)
+    (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())
@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Estimate the LOD0-2 infinite-worker floor from instrumented FI/TN tile timelines."""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+OUT = ROOT / "Saved/Linux/l83/serial-floor"
+DATA = ROOT / "Standalone/data/owner_world.json"
+BINARY = ROOT / "Saved/Linux/l83/profile-bin/VoxelForgeCoreHeadless-l83-profile-clang19"
+CAMERAS = {
+    1: (25.8369140625, 3.19921875, -737.0),
+    2: (19.166015625, -14.2001953125, -753.3740234375),
+}
+STRATE_NAMES = {1: "FI", 2: "TN"}
+EXPECTED_NEAR = {1: 1216, 2: 1344}
+EXPECTED_STAMP = "08af4e98c86a69477cc79e21463d8c93bcd11ebe0f077d7c11519031d9141fa9"
+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)
+SHAPE_FREE = re.compile(r"^\[ShapeReadiness\] shape_free_desired=(\d+)/(\d+) near_ring=(\d+)/27", re.M)
+STAMP = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=(\S+)$", re.M)
+FINISH = re.compile(
+    r"^\[CpuTileTimeline\] event=finish key=\((-?\d+),(-?\d+),(-?\d+),lod(\d+)\) "
+    r"at_ms=([0-9.]+).*?queued_ms=([0-9.]+) dispatch_ms=([0-9.]+) finish_ms=([0-9.]+) "
+    r"service_ms=([0-9.]+) worker_ms=([0-9.]+)$", re.M)
+
+
+def run(strate: int) -> dict:
+    OUT.mkdir(parents=True, exist_ok=True)
+    log = OUT / f"{STRATE_NAMES[strate].lower()}-w16.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=16", "--clip-radius=3",
+        "--max-clip-level=8", "--max-level=8", f"--camera={camera}",
+        "--profile-lod-work", "--instrument-tile-drain",
+    ]
+    started = time.monotonic()
+    with log.open("w", encoding="utf-8") as stream:
+        result = subprocess.run(command, cwd=ROOT, stdout=stream,
+                                stderr=subprocess.STDOUT, check=False, timeout=1740)
+    output = log.read_text(encoding="utf-8", errors="replace")
+    pipeline_match, near_match = PIPELINE.search(output), NEAR.search(output)
+    free_match, stamp_match = SHAPE_FREE.search(output), STAMP.search(output)
+    if result.returncode or pipeline_match is None or near_match is None:
+        raise RuntimeError(f"serial-floor run failed: exit={result.returncode} {log}")
+    if stamp_match is None or stamp_match.group(1) != EXPECTED_STAMP:
+        raise RuntimeError(f"source stamp mismatch in {log}")
+    pipeline = dict(FIELDS.findall(pipeline_match.group(1)))
+    near_done, near_total = int(near_match.group(1)), int(near_match.group(2))
+    if near_done != near_total or near_total != EXPECTED_NEAR[strate]:
+        raise RuntimeError(f"incomplete near tiles in {log}: {near_done}/{near_total}")
+    events = []
+    for match in FINISH.finditer(output):
+        lod = int(match.group(4))
+        if lod > 2:
+            continue
+        events.append({
+            "lod": lod, "finish_ms": float(match.group(5)), "queued_ms": float(match.group(6)),
+            "dispatch_ms": float(match.group(7)), "service_ms": float(match.group(9)),
+            "worker_ms": float(match.group(10)),
+        })
+    if len(events) != EXPECTED_NEAR[strate]:
+        raise RuntimeError(f"near timeline events incomplete in {log}: {len(events)}/{EXPECTED_NEAR[strate]}")
+    floors = [event["queued_ms"] + event["service_ms"] for event in events]
+    by_lod = {}
+    for lod in range(3):
+        subset = [event for event in events if event["lod"] == lod]
+        if subset:
+            by_lod[lod] = {
+                "tiles": len(subset),
+                "max_service_ms": max(event["service_ms"] for event in subset),
+                "max_wait_ms": max(event["dispatch_ms"] - event["queued_ms"] for event in subset),
+                "last_actual_finish_ms": max(event["finish_ms"] for event in subset),
+                "max_enqueue_plus_service_ms": max(e["queued_ms"] + e["service_ms"] for e in subset),
+            }
+    row = {
+        "strate": strate, "name": STRATE_NAMES[strate], "workers": 16,
+        "exit_code": result.returncode, "wall_s": round(time.monotonic() - started, 3),
+        "near_tiles": near_total, "lod02_ready_s": float(near_match.group(3)),
+        "shape_s": float(pipeline["shape"].removesuffix("s")),
+        "shape_free_desired": int(free_match.group(1)) if free_match else None,
+        "shape_free_near_ring": int(free_match.group(3)) if free_match else None,
+        "infinite_worker_floor_estimate_s": round(max(floors) / 1000.0, 6),
+        "max_near_service_ms": max(event["service_ms"] for event in events),
+        "last_near_enqueue_ms": max(event["queued_ms"] for event in events),
+        "max_near_enqueue_plus_service_ms": max(floors),
+        "max_near_queue_wait_ms": max(event["dispatch_ms"] - event["queued_ms"] for event in events),
+        "per_lod": by_lod, "source_stamp": stamp_match.group(1),
+        "log": str(log.relative_to(ROOT)),
+    }
+    print("[L83SerialFloor] " + json.dumps(row, sort_keys=True), flush=True)
+    return row
+
+
+def main() -> int:
+    for path in (BINARY, DATA):
+        if not path.is_file():
+            raise SystemExit(f"missing input: {path}")
+    rows = [run(strate) for strate in (1, 2)]
+    (OUT / "summary.json").write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())