test: prepare matched projected pair Taylor-Green repeat

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(