diff --git a/tests/validation/taylor_green/assess_projected_pairs.py b/tests/validation/taylor_green/assess_projected_pairs.py
new file mode 100644
index 0000000..8d02595
--- /dev/null
+++ b/tests/validation/taylor_green/assess_projected_pairs.py
@@ -0,0 +1,208 @@
+"""New-source four-case repeat of 11901a7; frozen oracle and gates are unchanged."""
+
+import argparse
+import json
+import math
+from pathlib import Path
+
+from assess import BASELINE_FACTOR, FLOORS
+from oracle import LIMITS, TIMES, U, rectangles
+from run_case import (INITIALIZATION, sample_steps, NUMERICAL_REVISION, SOURCE_SHA256,
+                      BUILD_CAPTURE_SHA256, REFERENCE_HARNESS)
+
+# Host-generated snapshots of a135252 at the frozen two local configurations.
+# These describe expected inputs; actual sidecars are captured before cleanup.
+BUILD_SOURCES = {
+    128: "803177416b7599ab89cc5afd35cf4a28a673b2f4bb8e8a6a1abc0b6d77610873",
+    256: "17ec5e60f87b840d3d13b8778bc8a3aecf485469c5a061af46c5cc19b73396cf",
+}
+CASES = ((100, 128, 1, "constant"), (100, 256, 1, "constant"),
+         (100, 128, 2, "constant"), (100, 128, 2, "projected_linear"))
+NUMERICS = dict(lattice_model="D2Q9", collision="SRT", store_precision="FP32",
+                ddf_shifting=True, streaming="double_buffer", les_enabled=False,
+                boundary="periodic", block_shape=[8, 8],
+                sample_phase="postcollision G at root synchronization")
+
+
+def finite(value):
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
+
+
+def finite_tree(value):
+    if isinstance(value, dict):
+        return all(finite_tree(v) for v in value.values())
+    if isinstance(value, list):
+        return all(finite_tree(v) for v in value)
+    return not isinstance(value, float) or math.isfinite(value)
+
+
+def at_steps(rows, steps, n):
+    return len(rows) == len(steps) and all(
+        type(r.get("step")) is int and r["step"] == step and finite(r.get("tstar"))
+        and abs(r["tstar"] - step * U / n) <= 3e-17
+        for r, step in zip(rows, steps)
+    )
+
+
+def maximum(rows, metric):
+    values = [r.get(metric) for r in rows]
+    return max(values) if values and all(finite(v) and v >= 0 for v in values) else None
+
+
+def valid_builds(record, n, depth):
+    builds = record.get("actual_build_sidecars", [])
+    if not isinstance(builds, list) or len(builds) != depth:
+        return False
+    for level, build in enumerate(builds):
+        if not isinstance(build, dict):
+            return False
+        size = n * 2**level
+        cfg = build.get("config", {})
+        expected = dict(nx=10, ny=10, nz=1, lattice_model="D2Q9", collision="SRT",
+                        store_precision="FP32", streaming="double_buffer", ddf_shifting=True,
+                        les_enabled=False, velocity=0.0, viscosity=U*size/100, rho=1.0,
+                        omega=1/(.5+3*U*size/100), threads_per_block=256)
+        ptx = build.get("ptx_sha256", "")
+        if (build.get("source_sha256") != BUILD_SOURCES[size]
+                or build.get("arch") != "sm_70" or build.get("flags") != ["--use_fast_math"]
+                or build.get("n_objects") != 0 or not isinstance(cfg, dict)
+                or any(cfg.get(k) != value for k, value in expected.items())
+                or not isinstance(ptx, str) or len(ptx) != 64
+                or any(c not in "0123456789abcdef" for c in ptx)):
+            return False
+    return True
+
+
+def assess(records, expected_revision=None):
+    indexed, duplicates = {}, []
+    for record in records:
+        key = tuple(record.get(k) for k in ("re_L", "n", "depth", "explosion"))
+        if key in indexed:
+            duplicates.append(str(key))
+        else:
+            indexed[key] = record
+    missing = [str(k) for k in CASES if k not in indexed]
+    checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(CASES),
+                  no_duplicate_cases=not duplicates, finite_receipts=finite_tree(records))
+    sources = [r.get("source", {}) for r in records]
+    checks["same_source_and_harness"] = bool(sources) and len({
+        json.dumps(s, sort_keys=True) for s in sources}) == 1
+    checks["clean_a135252_numerical_source"] = bool(sources) and all(
+        s.get("dirty") is False and s.get("source_tree_sha256") == SOURCE_SHA256
+        and bool(s.get("revision")) and bool(s.get("harness_sha256"))
+        and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
+        and (expected_revision is None or s.get("revision") == expected_revision) for s in sources)
+    runtimes = [r.get("runtime_identity") for r in records]
+    checks["same_recorded_runtime"] = bool(runtimes) and all(
+        isinstance(v, dict) and all(v.get(k) for k in
+            ("device", "nvcc_path", "nvcc_version", "python", "numpy", "pycuda"))
+        and isinstance(v["device"], dict) and v["device"].get("capability") == [7, 0]
+        and all(v["device"].get(k) for k in ("name", "pci_bus_id", "driver", "runtime"))
+        for v in runtimes
+    ) and len({json.dumps(v, sort_keys=True) for v in runtimes}) == 1
+    devices = [r.get("cuda_visible_devices") for r in records]
+    checks["same_explicit_device"] = bool(devices) and all(
+        isinstance(d, str) and d and "," not in d and d != "-1" for d in devices
+    ) and len(set(devices)) == 1
+    enabled = [r.get("early_pressure", {}).get("enabled") for r in records]
+    checks["same_early_sampling"] = bool(enabled) and all(type(v) is bool for v in enabled) and len(set(enabled)) == 1
+    shared_valid = all(checks.values())
+    complete, case_valid, early = {}, {}, []
+    for key in CASES:
+        if key not in indexed:
+            continue
+        r = indexed[key]
+        _, n, depth, scheme = key
+        label = f"{n}/{depth}/{scheme}"
+        rows = r.get("rows", [])
+        main_steps, early_steps = sample_steps(n, True)
+        complete[key] = r.get("completed") is True and at_steps(rows, main_steps, n)
+        checks[label + " complete_times"] = complete[key]
+        checks[label + " scales"] = (
+            r.get("U") == U and finite(r.get("viscosity"))
+            and abs(r["viscosity"] - U * n / 100) < 1e-15
+            and r.get("tau_by_level") == [0.5 + 3 * U * n / 100 * 2**l for l in range(depth)]
+            and r.get("mach") == U * math.sqrt(3)
+        )
+        checks[label + " actual_builds"] = valid_builds(r, n, depth)
+        checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION
+                                         and r.get("reference_harness") == REFERENCE_HARNESS)
+        checks[label + " numerics"] = r.get("numerics") == NUMERICS and r.get("initialization") == INITIALIZATION
+        runtime = r.get("runtime_source", {})
+        package = Path(runtime.get("checkout", "")) / "src/CelerisLab"
+        checks[label + " runtime_source"] = bool(runtime.get("checkout")) and all(
+            runtime.get(k) == str(package / tail) for k, tail in
+            (("module", "__init__.py"), ("compiler", "cuda/compiler_v2.py"), ("kernel_root", "lbm/kernels")))
+        expected_patch = [] if depth == 1 else rectangles(n, depth)
+        checks[label + " patch"] = json.dumps(r.get("refinements")) == json.dumps(expected_patch)
+        cells = [n*n] if depth == 1 else [8704, 30720]
+        hops = [16*n * 2**l for l in range(depth)]
+        work = r.get("workload", {})
+        checks[label + " workload"] = (
+            work.get("active_cells_by_level") == cells and work.get("level_hops") == hops
+            and work.get("root_intervals") == 16*n
+            and work.get("active_collision_updates_by_level") == [c*h for c, h in zip(cells, hops)]
+            and work.get("active_collision_updates") == sum(c*h for c, h in zip(cells, hops))
+        )
+        checks[label + " absolute"] = bool(rows) and all(
+            finite(row.get(metric)) and 0 <= row[metric] <= limit
+            for row in rows for metric, limit in LIMITS.items())
+        trace = r.get("early_pressure", {})
+        trace_rows = trace.get("rows", [])
+        peak = maximum(trace_rows, "pressure_l2")
+        trace_complete = at_steps(trace_rows, early_steps, n) and peak is not None
+        checks[label + " early_trace"] = (
+            trace_complete and trace.get("sampled_max_pressure_l2") == peak
+            and len(rows) >= 2 and all(trace_rows[i]["pressure_l2"] == rows[j].get("pressure_l2")
+                                      for i, j in ((0, 0), (-1, 1)))
+            if trace.get("enabled") is True else
+            trace.get("enabled") is False and trace_rows == [] and trace.get("sampled_max_pressure_l2") is None
+        )
+        early.append(dict(n=n, depth=depth, explosion=scheme, enabled=trace.get("enabled"),
+                          samples=len(trace_rows), complete=trace_complete,
+                          sampled_max_pressure_l2=peak))
+        case_valid[key] = all(value for name, value in checks.items() if name.startswith(label + " "))
+
+    def worst(key, metric):
+        return maximum(indexed[key]["rows"][1:], metric) if complete.get(key) else None
+
+    comparisons, scheme_results = [], {}
+    for scheme in ("constant", "projected_linear"):
+        key = (100, 128, 2, scheme)
+        for metric, floor in FLOORS.items():
+            amr, coarse, fine = (worst(k, metric) for k in (key, CASES[0], CASES[1]))
+            budget = None if coarse is None else max(BASELINE_FACTOR * coarse, floor)
+            budget = budget if finite(budget) else None
+            ratio = None if amr is None or fine is None else amr / max(fine, 1e-30)
+            passed = amr is not None and budget is not None and amr <= budget
+            checks[f"baseline {scheme}/{metric}"] = passed
+            comparisons.append(dict(explosion=scheme, metric=metric, amr=amr,
+                                    uniform128=coarse, baseline_budget=budget, passed=passed,
+                                    uniform256=fine, amr_to_uniform256=ratio if finite(ratio) else None))
+        baseline_passed = all(checks[f"baseline {scheme}/{metric}"] for metric in FLOORS)
+        valid = shared_valid and all(case_valid.get(k, False) for k in (CASES[0], CASES[1], key))
+        scheme_results[scheme] = dict(passed=valid and baseline_passed,
+                                     valid_case_and_controls=valid, baseline_passed=baseline_passed)
+    return dict(passed=all(checks.values()), missing=missing, duplicates=duplicates, checks=checks,
+                comparisons=comparisons, scheme_results=scheme_results, early_pressure=early,
+                numerical_revision=NUMERICAL_REVISION, source_tree_sha256=SOURCE_SHA256,
+                reference_harness=REFERENCE_HARNESS, expected_revision=expected_revision,
+                floors=FLOORS, baseline_factor=BASELINE_FACTOR,
+                main_statistic="maximum of four post-initial samples at t*=.12,.24,.36,.48",
+                early_statistic="sampled pressure maximum only, t*=k*.001875, k=0..64; not an all-step or continuous maximum",
+                finest_control="uniform256 and AMR/ uniform256 ratios are descriptive, not acceptance gates",
+                scope="2D periodic Re100 smooth-decay diagnostic; no new algorithm, convergence-order or general AMR qualification")
+
+
+if __name__ == "__main__":
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("files", type=Path, nargs="+")
+    p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--expected-revision", required=True)
+    args = p.parse_args()
+    if args.output.exists():
+        p.error("Output already exists; choose a new path to retain prior evidence")
+    result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision)
+    args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
+    print(json.dumps(result, indent=2, allow_nan=False))
+    raise SystemExit(0 if result["passed"] else 1)
diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py
index 9f0acb2..2f4954c 100644
--- a/tests/validation/taylor_green/run_case.py
+++ b/tests/validation/taylor_green/run_case.py
@@ -1,9 +1,10 @@
-"""One bounded GPU case. No production edits, plots, checkpoints or field dumps."""
+"""One bounded GPU case, with explicit opt-in experimental explosion."""
 
 import argparse
 import hashlib
 import json
 import os
+import shutil
 from pathlib import Path
 import subprocess
 import sys
@@ -12,6 +13,35 @@ import time
 import numpy as np
 from oracle import U, TIMES, LIMITS, initial, metrics, rectangles
 
+NUMERICAL_REVISION = "a135252ae78b47b57d7e77d19001700b3bedac8b"
+SOURCE_SHA256 = "a5dbfd804befce868067d79a23e5fd47dce7f505b3f9b1bb2145fb57cb5f2178"
+BUILD_CAPTURE_SHA256 = "3dbaa6547cb0a4cb1541054f3b64ad8b6e18cc70527ac6166e56b944e9bd464a"
+REFERENCE_HARNESS = "11901a7f77557a901e017a171eeae1758d584fa5"
+INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p"
+
+
+def sample_steps(n, early_pressure=False):
+    """Root synchronization steps; early ladder matches the 707afe6 diagnostic."""
+    main = [round(tstar * n / U) for tstar in TIMES]
+    early = list(range(0, 4 * n + 1, n // 16)) if early_pressure else []
+    return main, early
+
+
+def runtime_source(repo):
+    """Fail before CUDA if Python or the compiler snapshot comes from another tree."""
+    import CelerisLab
+    from CelerisLab.cuda import compiler_v2
+
+    paths = dict(module=Path(CelerisLab.__file__).resolve(),
+                 compiler=Path(compiler_v2.__file__).resolve(),
+                 kernel_root=Path(compiler_v2.kernel_path("blocks.cu")).resolve().parent)
+    package = repo / "src/CelerisLab"
+    expected = dict(module=package / "__init__.py", compiler=package / "cuda/compiler_v2.py",
+                    kernel_root=package / "lbm/kernels")
+    if paths != expected:
+        raise RuntimeError("Wrong imported source or CUDA source/include root")
+    return dict(checkout=str(repo), **{key: str(path) for key, path in paths.items()})
+
 
 def provenance(repo):
     git = lambda *a: subprocess.check_output(
@@ -34,6 +64,46 @@ def provenance(repo):
     )
 
 
+def verify_source(repo, revision):
+    """Exact clean harness commit, fixed production tree, and reviewed capture helper."""
+    import oracle
+    from tests.validation.multigrid import check_projected_pairs as gate
+
+    harness = repo / "tests/validation/taylor_green"
+    if (Path(__file__).resolve() != harness / "run_case.py"
+            or Path(oracle.__file__).resolve() != harness / "oracle.py"):
+        raise RuntimeError("Wrong runner or oracle checkout")
+    helper = repo / "tests/validation/multigrid/check_projected_pairs.py"
+    if (Path(gate.__file__).resolve() != helper
+            or hashlib.sha256(helper.read_bytes()).hexdigest() != BUILD_CAPTURE_SHA256):
+        raise RuntimeError("Wrong build-capture helper")
+    if gate.preflight(repo, revision, "candidate") != NUMERICAL_REVISION:
+        raise RuntimeError("Wrong production revision")
+    result = provenance(repo)
+    if result["source_tree_sha256"] != SOURCE_SHA256:
+        raise RuntimeError("Wrong numerical source fingerprint")
+    result["build_capture_sha256"] = BUILD_CAPTURE_SHA256
+    return result
+
+
+def build_capture(report, save):
+    from tests.validation.multigrid.check_projected_pairs import capture_builds
+    return capture_builds(report, save)
+
+
+def runtime_identity(ctx):
+    import pycuda
+    import pycuda.driver as cuda
+
+    nvcc = shutil.which("nvcc") or "/usr/local/cuda/bin/nvcc"
+    return dict(device=dict(name=ctx.device.name(), pci_bus_id=ctx.device.pci_bus_id(),
+                            capability=list(ctx.device.compute_capability()),
+                            driver=cuda.get_driver_version(), runtime=list(cuda.get_version())),
+                nvcc_path=str(Path(nvcc).resolve()),
+                nvcc_version=subprocess.check_output([nvcc, "--version"], text=True).strip(),
+                python=sys.version, numpy=np.__version__, pycuda=pycuda.VERSION_TEXT)
+
+
 def config(n, re):
     from CelerisLab.config import LBMConfig
 
@@ -71,19 +141,31 @@ def workload(masks, base_steps):
 def main():
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument("--repo", type=Path, required=True)
+    p.add_argument("--expected-revision", required=True)
+    p.add_argument("--run-gpu", action="store_true")
+    p.add_argument("--dry-run", action="store_true")
     p.add_argument("--re", type=int, choices=[100, 1000], required=True)
     p.add_argument("--n", type=int, choices=[32, 64, 128, 256], required=True)
     p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True)
     p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
+    p.add_argument("--early-pressure", action="store_true",
+                   help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
     args = p.parse_args()
-    if not os.environ.get("CUDA_VISIBLE_DEVICES"):
-        p.error("Explicit physical CUDA_VISIBLE_DEVICES selection required")
+    if args.output.exists():
+        p.error("Output already exists; choose a new path to retain prior evidence")
+    if args.explosion != "constant" and args.depth != 2:
+        p.error("projected_linear requires depth 2")
+    selected = os.environ.get("CUDA_VISIBLE_DEVICES", "")
+    if not args.dry_run and (not args.run_gpu or not selected or "," in selected or selected == "-1"):
+        p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
     repo = args.repo.resolve()
-    sys.path.insert(0, str(repo / "src"))
-    import CelerisLab
-
-    if Path(CelerisLab.__file__).resolve() != repo / "src/CelerisLab/__init__.py":
-        raise RuntimeError("Wrong imported source checkout")
+    sys.path[:0] = [str(repo / "src"), str(repo)]
+    source = verify_source(repo, args.expected_revision)
+    runtime = runtime_source(repo)
+    if args.dry_run:
+        print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True)))
+        return 0
     from CelerisLab.cuda.context import CudaContext
     from CelerisLab.lbm.blocks import BlockLayout2D
     from CelerisLab.lbm.block_stepper import PeriodicBlockStepper
@@ -96,9 +178,17 @@ def main():
     h = build_hierarchy(cfg, refinements)
     np.testing.assert_array_equal(leaf_partition(h), np.ones((args.n, args.n)))
     masks = [k == ACTIVE for k in h.kinds]
+    main_steps, early_steps = sample_steps(args.n, args.early_pressure)
     report = dict(
         schema=1,
-        source=provenance(repo),
+        source=source,
+        numerical_revision=NUMERICAL_REVISION,
+        reference_harness=REFERENCE_HARNESS,
+        actual_build_sidecars=[],
+        runtime_identity=None,
+        runtime_source=runtime,
+        historical_baseline_revision="9805383f9c50d867e8555471baf5dbda8333a652",
+        explosion=args.explosion,
         n=args.n,
         re_L=args.re,
         depth=args.depth,
@@ -107,21 +197,35 @@ def main():
         mach=U * np.sqrt(3),
         viscosity=cfg.viscosity,
         tau_by_level=[0.5 + 3 * c.viscosity for c in h.configs],
-        initialization="physical FP64 equilibrium G at actual cell centers; rho=1+3p",
-        cuda_visible_devices=os.environ["CUDA_VISIBLE_DEVICES"],
+        initialization=INITIALIZATION,
+        numerics=dict(lattice_model=cfg.lattice_model, collision=cfg.collision,
+                      store_precision=cfg.store_precision, ddf_shifting=cfg.ddf_shifting,
+                      streaming=cfg.streaming, les_enabled=cfg.les_enabled,
+                      boundary="periodic", block_shape=[8, 8],
+                      sample_phase="postcollision G at root synchronization"),
+        cuda_visible_devices=selected,
         device_id=0,
         limits=LIMITS,
         workload=workload(masks, round(TIMES[-1] * args.n / U)),
         rows=[],
+        early_pressure=dict(enabled=args.early_pressure, rows=[],
+                            sampled_max_pressure_l2=None),
         completed=False,
     )
     args.output.parent.mkdir(parents=True, exist_ok=True)
-    save = lambda: args.output.write_text(
-        json.dumps(report, indent=2, allow_nan=False) + "\n"
-    )
+    def save():
+        temporary = args.output.with_suffix(args.output.suffix + ".tmp")
+        text = json.dumps(report, indent=2, allow_nan=False) + "\n"
+        if len(text.encode()) > 1024 * 1024:
+            raise RuntimeError("Scalar receipt exceeds 1 MiB cap")
+        temporary.write_text(text)
+        temporary.replace(args.output)
+
     save()
     started = time.monotonic()
-    with CudaContext(0) as ctx:
+    with CudaContext(0) as ctx, build_capture(report, save):
+        report["runtime_identity"] = runtime_identity(ctx)
+        save()
         if args.depth == 1:
             layout = BlockLayout2D.build(args.n, args.n, 8, 8, periodic=(True, True))
             run = PeriodicBlockStepper(cfg, layout, context=ctx)
@@ -132,6 +236,7 @@ def main():
                 coarse_boundary="periodic",
                 refinements=refinements,
                 level_blocks=[(8, 8)] * args.depth,
+                explosion=args.explosion,
             )
         with run:
             run.initialize(
@@ -141,24 +246,30 @@ def main():
                 ]
             )
             initial_mass = None
-            for tstar in TIMES:
-                step = round(tstar * args.n / U)
+            for step in sorted(set(main_steps + early_steps)):
                 run.step(step - run.step_count)
                 arrays = (run.download(),) if args.depth == 1 else run.download()
                 row = metrics(arrays, masks, step, args.n, cfg.viscosity, initial_mass)
+                if not all(np.isfinite(value) for value in row.values()):
+                    raise FloatingPointError("nonfinite sample diagnostics")
                 row["step"] = step
                 if initial_mass is None:
                     initial_mass = row["mass"]
-                report["rows"].append(row)
+                if step in main_steps:
+                    report["rows"].append(row)
+                if step in early_steps:
+                    trace = report["early_pressure"]
+                    trace["rows"].append({key: row[key] for key in ("step", "tstar", "pressure_l2")})
+                    trace["sampled_max_pressure_l2"] = max(r["pressure_l2"] for r in trace["rows"])
                 save()
                 print(
                     json.dumps(
                         dict(
-                            event="PROGRESS",
+                            event="PROGRESS" if step in main_steps else "EARLY_PRESSURE",
                             n=args.n,
                             re_L=args.re,
                             depth=args.depth,
-                            **row,
+                            **(row if step in main_steps else report["early_pressure"]["rows"][-1]),
                         )
                     ),
                     flush=True,
@@ -168,7 +279,9 @@ def main():
                     # This is the first measured trajectory step, not discarded warmup.
                     run.step(1)
                     probe = (run.download(),) if args.depth == 1 else run.download()
-                    metrics(probe, masks, 1, args.n, cfg.viscosity, initial_mass)
+                    probe_metrics = metrics(probe, masks, 1, args.n, cfg.viscosity, initial_mass)
+                    if not all(np.isfinite(value) for value in probe_metrics.values()):
+                        raise FloatingPointError("nonfinite READY diagnostics")
                     report["ready_after_step"] = 1
                     save()
                     print(
diff --git a/tests/validation/taylor_green/test_projected_pairs.py b/tests/validation/taylor_green/test_projected_pairs.py
new file mode 100644
index 0000000..50b9f8b
--- /dev/null
+++ b/tests/validation/taylor_green/test_projected_pairs.py
@@ -0,0 +1,434 @@
+"""Host geometry, receipt-corruption tests and mocked scheduling; no CUDA/flow."""
+
+from contextlib import nullcontext, redirect_stdout
+from copy import deepcopy
+from dataclasses import asdict, replace
+import hashlib
+import io
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+import numpy as np
+
+import run_case
+from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
+                                    SOURCE_SHA256, assess)
+from oracle import LIMITS, TIMES, U, rectangles
+
+REPO = Path(__file__).resolve().parents[3]
+OLD_REVISION = "519473a61c7b319183aa54e55b9f097499dde0bf"
+MOCK_REVISION = "f" * 40
+RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0],
+                           driver=12040, runtime=[12, 4, 0]), nvcc_path="/mock/nvcc",
+               nvcc_version="mock nvcc", python="mock Python", numpy=np.__version__, pycuda="mock")
+
+
+def source():
+    return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256,
+                harness_sha256={"run_case.py": "a" * 64},
+                build_capture_sha256=run_case.BUILD_CAPTURE_SHA256)
+
+
+def build_metadata(size):
+    cfg = replace(run_case.config(size, 100), nx=10, ny=10)
+    cfg.validate()
+    return dict(source_sha256=BUILD_SOURCES[size], config=asdict(cfg), n_objects=0,
+                arch="sm_70", flags=["--use_fast_math"], ptx_sha256="b" * 64)
+
+
+def records(early=False):
+    from CelerisLab.lbm.hierarchy import build_hierarchy
+    from CelerisLab.lbm.refinement import ACTIVE
+
+    out = []
+    for re, n, depth, scheme in CASES:
+        cfg = run_case.config(n, re)
+        refinements = [] if depth == 1 else rectangles(n, depth)
+        h = build_hierarchy(cfg, refinements)
+        main_steps, early_steps = run_case.sample_steps(n, early)
+        out.append(dict(
+            re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
+            viscosity=cfg.viscosity, tau_by_level=[0.5+3*c.viscosity for c in h.configs],
+            refinements=refinements, numerics=NUMERICS.copy(), initialization=run_case.INITIALIZATION,
+            source=source(), numerical_revision=NUMERICAL_REVISION,
+            reference_harness=run_case.REFERENCE_HARNESS,
+            actual_build_sidecars=[build_metadata(n * 2**level) for level in range(depth)],
+            runtime_identity=deepcopy(RUNTIME), cuda_visible_devices="0",
+            runtime_source=run_case.runtime_source(REPO), completed=True,
+            workload=run_case.workload([k == ACTIVE for k in h.kinds], 16*n),
+            rows=[dict(step=s, tstar=s*U/n, **{k: 0.0 for k in LIMITS}) for s in main_steps],
+            early_pressure=dict(enabled=early, rows=[
+                dict(step=s, tstar=s*U/n, pressure_l2=.01 if s not in main_steps else 0.0) for s in early_steps],
+                sampled_max_pressure_l2=.01 if early else None)))
+    return out
+
+
+class ProjectedPairsTests(unittest.TestCase):
+    def test_exact_geometry_work_and_samples(self):
+        from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition
+
+        data = records(True)
+        self.assertEqual(sum(r["workload"]["active_collision_updates"] for r in data), 589299712)
+        for r in data:
+            n, depth = r["n"], r["depth"]
+            h = build_hierarchy(run_case.config(n, 100), r["refinements"])
+            np.testing.assert_array_equal(leaf_partition(h), np.ones((n, n)))
+            main, early = run_case.sample_steps(n, True)
+            self.assertEqual(main, [0, 4*n, 8*n, 12*n, 16*n])
+            self.assertEqual(early, [k*(n//16) for k in range(65)])
+            np.testing.assert_allclose(np.array(early)*U/n, np.arange(65)*.001875, rtol=0, atol=3e-17)
+            if depth == 2:
+                self.assertEqual(r["refinements"], [[(16, 16, 112, 96)]])
+        self.assertTrue(assess(data)["passed"])
+        self.assertTrue(assess(records())["passed"])
+
+    def test_exact_old519_geometry_and_table_parity(self):
+        from CelerisLab.lbm.blocks import BlockLayout2D
+        from CelerisLab.lbm.hierarchy import build_hierarchy
+        from CelerisLab.lbm.refinement_tables import projected_explosion_tables
+
+        # Frozen from executing exact 519473a61c7b319183aa54e55b9f097499dde0bf
+        # geometry and table definitions, after checking unchanged config,
+        # descriptors and blocks. No historical Git object is needed at runtime.
+        # Historical source SHA256, refinement.py then refinement_tables.py:
+        # 08c2a0557276a98cbcaa459d9bb7a9735d8bbbf64a6114c57af2fda61481a687
+        # dabff948bb30496384d1a52c666b0edbc2085f9c61ed1bbc39f748502629113d
+        frozen = {
+            "covered": ("|b1", (128, 128), "14a1b5be54e3309bea825a00a653105951f0c805622b877a86dd453d65556df3"),
+            "fine_kind": ("|u1", (256, 256), "891ff72c3d60686c62b1f35ffe59d054ac23b8f9d59aeb5782bc941202794cc0"),
+            "ghost_fine": ("<u8", (2880,), "2d0311ac0feb1c0e1e2b6bcf806fe616b8217513de218407267dcd7652287579"),
+            "ghost_coarse": ("<u4", (2880,), "5a3b305a0f841c8d05a13f105fda02b43ef875ab1321f9bca579f2b8ae313233"),
+            "coalesce_coarse": ("<u8", (1060,), "a1d88c60f282153dcf2bb497c59d777968b6db773d061d8fea08163c98155967"),
+            "coalesce_fine": ("<u8", (1060, 4), "08914c5d17c1ec8e2bfd234f581e4b3283de26296ae7d80403385feed7b89532"),
+            "neighbors": ("<u8", (2880, 4), "cfef3a07103241e3dfff24fd65ab967afcf20520f699574b191044d47cf5d1f4"),
+            "offsets": ("<f4", (2880, 2), "c98ba658d5f2cc18605e029fbe1da448ded3fd2436edcaf8ea097fa46aeb1f2d"),
+        }
+        hierarchy = build_hierarchy(run_case.config(128, 100), rectangles(128, 2))
+        pair = hierarchy.pairs[0]
+        self.assertEqual((pair.coarse_shape, pair.fine_shape, pair.fine_origin),
+                         ((128, 128), (256, 256), (0.0, 0.0)))
+        self.assertEqual(len(np.unique(pair.ghost_coarse)), 720)
+        self.assertTrue((hierarchy.kinds[0].ravel()[pair.ghost_coarse] == 0).all())
+        layout = BlockLayout2D.build(128, 128, 8, 8, periodic=(True, True))
+        neighbors, offsets = projected_explosion_tables(pair, layout, hierarchy.kinds[0])
+        arrays = dict(neighbors=neighbors, offsets=offsets)
+        arrays.update({name: getattr(pair, name) for name in frozen if name not in arrays})
+        for name, array in arrays.items():
+            with self.subTest(array=name):
+                self.assertEqual((array.dtype.str, array.shape, hashlib.sha256(array.tobytes()).hexdigest()),
+                                 frozen[name])
+
+    def test_actual_build_capture_rejects_tampered_ptx_and_sidecar(self):
+        from CelerisLab.cuda import compiler_v2
+        from tests.validation.multigrid.check_projected_pairs import capture_builds
+
+        for tamper in (None, "ptx", "sidecar"):
+            with self.subTest(tamper=tamper), tempfile.TemporaryDirectory() as tmp:
+                path = Path(tmp) / "blocks.ptx"
+                path.write_bytes(b"host-only mock PTX")
+                metadata = dict(ptx_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
+                if tamper == "ptx":
+                    path.write_bytes(b"tampered")
+                if tamper == "sidecar":
+                    metadata["ptx_sha256"] = "0" * 64
+                Path(str(path) + ".json").write_text(json.dumps(metadata))
+                report = {"actual_build_sidecars": []}
+                with patch.object(compiler_v2, "load_module", return_value="mock-module") as load:
+                    with capture_builds(report, lambda: None):
+                        if tamper:
+                            with self.assertRaisesRegex(AssertionError, "PTX and build sidecar"):
+                                compiler_v2.load_module(str(path))
+                            load.assert_not_called()
+                            self.assertEqual(report["actual_build_sidecars"], [])
+                        else:
+                            self.assertEqual(compiler_v2.load_module(str(path)), "mock-module")
+                            self.assertEqual(report["actual_build_sidecars"], [metadata])
+                    self.assertIs(compiler_v2.load_module, load)
+
+    def test_frozen_formula_and_finest_control(self):
+        data = records(True)
+        data[0]["rows"][2]["velocity_l2"] = .002
+        data[0]["rows"][0]["velocity_l2"] = .04  # retained absolute gate, excluded from worst
+        for r in data[2:]:
+            for row in r["rows"][1:]:
+                row.update(velocity_l2=.0039, pressure_l2=.02, energy_relative=.004)
+            r["early_pressure"]["rows"][-1]["pressure_l2"] = .02
+            r["early_pressure"]["sampled_max_pressure_l2"] = .02
+        data[1]["rows"][2]["velocity_l2"] = .00001
+        data[3]["early_pressure"]["rows"][7]["pressure_l2"] = .9
+        data[3]["early_pressure"]["sampled_max_pressure_l2"] = .9
+        result = assess(data)
+        self.assertTrue(result["passed"])
+        velocity = result["comparisons"][0]
+        self.assertEqual(velocity["baseline_budget"], .004)
+        self.assertAlmostEqual(velocity["amr_to_uniform256"], 390)
+        self.assertEqual(result["early_pressure"][-1]["sampled_max_pressure_l2"], .9)
+        data[3]["rows"][-1]["pressure_l2"] = .02001
+        result = assess(data)
+        self.assertFalse(result["passed"])
+        self.assertEqual(result["comparisons"][4]["amr"], .02001)
+
+    def test_scheme_verdicts_are_independent(self):
+        data = records()
+        data[2]["rows"][-1]["pressure_l2"] = .03
+        result = assess(data)
+        self.assertFalse(result["passed"])
+        self.assertFalse(result["scheme_results"]["constant"]["passed"])
+        self.assertTrue(result["scheme_results"]["projected_linear"]["passed"])
+
+    def test_corruptions_cannot_pass(self):
+        good = records(True)
+        corruptions = [
+            lambda r: r.update(completed=False),
+            lambda r: r["rows"].pop(),
+            lambda r: r["rows"][-1].update(pressure_l2=float("nan")),
+            lambda r: r["rows"][-1].update(energy_relative=float("inf")),
+            lambda r: r["rows"][-1].update(pressure_l2=1e308),
+            lambda r: r["rows"][-1].update(velocity_l2=-1),
+            lambda r: r["rows"][-1].update(step=2047),
+            lambda r: r["rows"][-1].update(tstar=.4801),
+            lambda r: r["source"].update(dirty=True),
+            lambda r: r["source"].update(source_tree_sha256="wrong"),
+            lambda r: r["source"].update(revision=OLD_REVISION),
+            lambda r: r["source"].pop("build_capture_sha256"),
+            lambda r: r.update(numerical_revision=OLD_REVISION),
+            lambda r: r.update(reference_harness="wrong"),
+            lambda r: r.pop("runtime_identity"),
+            lambda r: r["runtime_identity"].pop("nvcc_version"),
+            lambda r: r["runtime_identity"].update(device="wrong"),
+            lambda r: r["runtime_identity"]["device"].update(capability=[8, 0]),
+            lambda r: r.update(cuda_visible_devices="0,1"),
+            lambda r: r.pop("actual_build_sidecars"),
+            lambda r: r["actual_build_sidecars"].pop(),
+            lambda r: r["actual_build_sidecars"].reverse(),
+            lambda r: r["actual_build_sidecars"][0].update(source_sha256="c" * 64),
+            lambda r: r["actual_build_sidecars"][0].update(arch="sm_80"),
+            lambda r: r["actual_build_sidecars"][0].update(flags=[]),
+            lambda r: r["actual_build_sidecars"][0].update(ptx_sha256="not-a-hash"),
+            lambda r: r["actual_build_sidecars"][0]["config"].update(ddf_shifting=False),
+            lambda r: r["source"]["harness_sha256"].update(extra="mixed"),
+            lambda r: r.update(viscosity=.01),
+            lambda r: r["tau_by_level"].reverse(),
+            lambda r: r["numerics"].update(ddf_shifting=False),
+            lambda r: r.update(initialization="different equilibrium"),
+            lambda r: r.update(refinements=[[(16, 16, 112, 104)]]),
+            lambda r: r["workload"].update(active_collision_updates=0),
+            lambda r: r["runtime_source"].update(kernel_root="/wrong/installed/tree"),
+            lambda r: r["early_pressure"]["rows"].pop(),
+            lambda r: r["early_pressure"]["rows"][8].update(step=65),
+            lambda r: r["early_pressure"]["rows"][-1].update(pressure_l2=.005),
+            lambda r: r["early_pressure"]["rows"][8].update(pressure_l2=float("nan")),
+            lambda r: r["early_pressure"].update(sampled_max_pressure_l2=.5),
+            lambda r: r["early_pressure"].update(enabled=False),
+        ]
+        for mutation in corruptions:
+            bad = deepcopy(good)
+            mutation(bad[-1])
+            with self.subTest(mutation=mutation):
+                result = assess(bad)
+                self.assertFalse(result["passed"])
+                json.dumps(result, allow_nan=False)  # failure summary itself remains serializable
+        self.assertTrue(assess(good, expected_revision=MOCK_REVISION)["passed"])
+        self.assertFalse(assess(good, expected_revision=OLD_REVISION)["passed"])
+        old = deepcopy(good)
+        for record in old:
+            record["source"].update(revision=OLD_REVISION,
+                source_tree_sha256="766e173df81d2881d3871007f15d5e321f5c7205ae578b0f80b43d1e7fe3e220")
+        self.assertFalse(assess(old)["passed"])
+        self.assertFalse(assess(good[:-1])["passed"])
+        self.assertFalse(assess(good + [good[-1]])["passed"])
+        self.assertFalse(assess([])["passed"])
+        incomplete = deepcopy(good)
+        incomplete[-1]["completed"] = False
+        self.assertFalse(assess(incomplete)["comparisons"][3]["passed"])
+
+    def test_pinned_build_sources_match_host_generated_inputs(self):
+        from CelerisLab.cuda import compiler_v2
+
+        for size in (128, 256):
+            cfg = replace(run_case.config(size, 100), nx=10, ny=10)
+            cfg.validate()
+            root = compiler_v2.generate_config(cfg)
+            try:
+                actual = json.loads((Path(root) / ".celeris-build").read_text())
+                expected = build_metadata(size)
+                self.assertEqual(actual, {key: expected[key] for key in
+                                          ("source_sha256", "config", "n_objects")})
+            finally:
+                compiler_v2.cleanup_build_dir(root)
+
+    def test_source_preflight_requires_clean_exact_revision_and_tree(self):
+        import oracle
+        from tests.validation.multigrid import check_projected_pairs as gate
+
+        for module in (run_case, oracle):
+            with self.subTest(imported_module=module.__name__), \
+                 patch.object(module, "__file__", "/wrong/checkout/" + Path(module.__file__).name), \
+                 patch.object(gate, "preflight") as preflight:
+                with self.assertRaisesRegex(RuntimeError, "Wrong runner or oracle checkout"):
+                    run_case.verify_source(REPO, MOCK_REVISION)
+                preflight.assert_not_called()
+        for expected, status, diff, fingerprint in (
+                (MOCK_REVISION, "", 0, SOURCE_SHA256),
+                (OLD_REVISION, "", 0, SOURCE_SHA256),
+                (MOCK_REVISION, " M tracked.py", 0, SOURCE_SHA256),
+                (MOCK_REVISION, "", 1, SOURCE_SHA256),
+                (MOCK_REVISION, "", 0, "wrong")):
+            provenance = dict(source(), source_tree_sha256=fingerprint)
+            with self.subTest(expected=expected, status=status, diff=diff, fingerprint=fingerprint), \
+                 patch.object(gate.subprocess, "check_output", side_effect=[MOCK_REVISION, status]), \
+                 patch.object(gate.subprocess, "call", return_value=diff), \
+                 patch.object(run_case, "provenance", return_value=provenance):
+                if expected != MOCK_REVISION or status or diff or fingerprint != SOURCE_SHA256:
+                    with self.assertRaises((ValueError, RuntimeError)):
+                        run_case.verify_source(REPO, expected)
+                else:
+                    self.assertEqual(run_case.verify_source(REPO, expected), source())
+
+    def test_explicit_opt_in_selector_and_dry_run_without_cuda(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            output = Path(tmp) / "case.json"
+            base = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
+                    "--depth", "1", "--output", str(output)]
+            pinned = base + ["--expected-revision", MOCK_REVISION]
+            for argv, device in ((base + ["--run-gpu"], "0"), (pinned, "0"),
+                    (pinned + ["--run-gpu"], ""), (pinned + ["--run-gpu"], "0,1"),
+                    (pinned + ["--run-gpu"], "-1"),
+                    (pinned + ["--run-gpu", "--explosion", "projected"], "0"),
+                    (pinned + ["--run-gpu", "--explosion", "projected_linear"], "0")):
+                with self.subTest(argv=argv, device=device), patch.object(sys, "argv", argv), \
+                     patch.dict("os.environ", CUDA_VISIBLE_DEVICES=device), \
+                     patch.object(run_case, "verify_source") as verify, patch("sys.stderr", io.StringIO()):
+                    with self.assertRaises(SystemExit) as caught:
+                        run_case.main()
+                    self.assertEqual(caught.exception.code, 2)
+                    verify.assert_not_called()
+                    self.assertFalse(output.exists())
+            stdout = io.StringIO()
+            cuda_modules = {name: None for name in ("pycuda", "CelerisLab.cuda.context",
+                           "CelerisLab.lbm.block_stepper", "CelerisLab.lbm.refinement_stepper")}
+            with patch.object(sys, "argv", pinned + ["--dry-run"]), \
+                 patch.dict(sys.modules, cuda_modules), patch.dict("os.environ", CUDA_VISIBLE_DEVICES=""), \
+                 patch.object(run_case, "verify_source", return_value=source()) as verify, \
+                 patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
+                 redirect_stdout(stdout):
+                self.assertEqual(run_case.main(), 0)
+            verify.assert_called_once_with(REPO, MOCK_REVISION)
+            self.assertTrue(json.loads(stdout.getvalue())["dry_run"])
+            self.assertFalse(output.exists())
+
+    def test_runtime_rejects_installed_kernel_tree(self):
+        from CelerisLab.cuda import compiler_v2
+        with patch.object(compiler_v2, "_KERNEL_DIR", "/other/CelerisLab/lbm/kernels"):
+            with self.assertRaisesRegex(RuntimeError, "CUDA source/include root"):
+                run_case.runtime_source(REPO)
+
+    def test_existing_output_is_never_overwritten(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            output = Path(tmp) / "failed.json"
+            output.write_text('{"completed": false}\n')
+            argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
+                    "--depth", "1", "--output", str(output),
+                    "--expected-revision", MOCK_REVISION, "--run-gpu"]
+            with patch.object(sys, "argv", argv), patch("sys.stderr", io.StringIO()):
+                with self.assertRaises(SystemExit) as caught:
+                    run_case.main()
+            self.assertEqual(caught.exception.code, 2)
+            self.assertEqual(output.read_text(), '{"completed": false}\n')
+            result = subprocess.run([sys.executable, str(Path(__file__).with_name("assess_projected_pairs.py")),
+                                     "unused.json", "--expected-revision", MOCK_REVISION,
+                                     "--output", str(output)], capture_output=True)
+            self.assertEqual(result.returncode, 2)
+            self.assertEqual(output.read_text(), '{"completed": false}\n')
+
+    def test_mocked_runner_default_early_and_failure(self):
+        # Replace the GPU context and steppers before import. No CUDA module or flow is used.
+        for early, depth, fail in ((False, 1, None), (False, 2, None), (True, 2, None),
+                                   (True, 2, "resource"), (True, 2, "nonfinite"),
+                                   (True, 2, "probe_nonfinite")):
+            calls, selectors = [], []
+
+            class FakeRun:
+                def __init__(self, *args, **kwargs):
+                    self.step_count = 0
+                    if "coarse_boundary" in kwargs:
+                        selectors.append(kwargs["explosion"])
+
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, *args):
+                    pass
+
+                def initialize(self, *arrays):
+                    pass
+
+                def step(self, count):
+                    self.assertion = count >= 0
+                    if not self.assertion:
+                        raise AssertionError("backward step")
+                    if fail == "resource" and self.step_count >= 16:
+                        raise MemoryError("mock resource failure")
+                    self.step_count += count
+                    calls.append(self.step_count)
+
+                def download(self):
+                    return None
+
+            modules = {
+                "CelerisLab.cuda.context": SimpleNamespace(CudaContext=FakeRun),
+                "CelerisLab.lbm.block_stepper": SimpleNamespace(PeriodicBlockStepper=FakeRun),
+                "CelerisLab.lbm.refinement_stepper": SimpleNamespace(FixedPatchStepper=FakeRun),
+            }
+
+            def metric(arrays, masks, step, n, nu, mass):
+                return dict(step=step, tstar=step*U/n, mass=16384.,
+                            **{k: float("nan") if ((fail == "nonfinite" and step == 16)
+                               or (fail == "probe_nonfinite" and step == 1))
+                               and k == "energy_relative" else 0.0 for k in LIMITS})
+
+            with tempfile.TemporaryDirectory() as tmp:
+                output = Path(tmp) / "case.json"
+                argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
+                        "--depth", str(depth), "--output", str(output),
+                        "--expected-revision", MOCK_REVISION, "--run-gpu"]
+                if early:
+                    argv += ["--early-pressure", "--explosion", "projected_linear"]
+                with patch.dict(sys.modules, modules), patch.object(sys, "argv", argv), \
+                     patch.dict("os.environ", CUDA_VISIBLE_DEVICES="mock"), \
+                     patch.object(run_case, "verify_source", return_value=source()) as verify, \
+                     patch.object(run_case, "runtime_identity", return_value=deepcopy(RUNTIME)), \
+                     patch.object(run_case, "build_capture", return_value=nullcontext()), \
+                     patch.object(run_case, "metrics", side_effect=metric), redirect_stdout(io.StringIO()):
+                    if fail:
+                        with self.assertRaises(MemoryError if fail == "resource" else FloatingPointError):
+                            run_case.main()
+                    else:
+                        self.assertEqual(run_case.main(), 0)
+                receipt = json.loads(output.read_text())
+                verify.assert_called_once_with(REPO, MOCK_REVISION)
+                self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else [])
+                self.assertEqual(receipt["source"], source())
+                self.assertEqual(receipt["runtime_identity"], RUNTIME)
+                self.assertEqual(receipt["completed"], not fail)
+                self.assertEqual(calls[:2], [0, 1])  # READY probe is part of this trajectory
+                if not fail:
+                    self.assertEqual([r["tstar"] for r in receipt["rows"]], TIMES)
+                    self.assertEqual(len(receipt["early_pressure"]["rows"]), 65 if early else 0)
+                    self.assertEqual(calls[-1], 2048)
+                else:
+                    self.assertEqual(len(receipt["rows"]), 1)
+                    self.assertEqual(len(receipt["early_pressure"]["rows"]),
+                                     {"resource": 3, "nonfinite": 2, "probe_nonfinite": 1}[fail])
+                    if fail == "probe_nonfinite":
+                        self.assertNotIn("ready_after_step", receipt)
+
+
+if __name__ == "__main__":
+    unittest.main()