test: retain independent Jet CV continuation measurements

tests/unit/test_jet_cv_measurements.py

@@ -0,0 +1,182 @@
+"""Manufactured references for continuation instrumentation, no CUDA."""
+
+import sys
+from pathlib import Path
+import tempfile
+import unittest
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
+from jet_cv_measurements import (
+    sample_steps,
+    factors,
+    ray_geometry,
+    reconstruct_aperture,
+    contour_terms,
+    LINEAR,
+    QUADRATIC,
+    OFFSETS,
+    MeasurementRecorder,
+)
+from run_jet_circle_cv import (
+    case,
+    CX,
+    CY,
+    W,
+    cv,
+    aperture_probe,
+    measured_aperture_flux,
+)
+from analyze_jet_cv_continuation import integrate, analyze
+
+
+class Measurements(unittest.TestCase):
+    def test_common_physical_schedule(self):
+        a = sample_steps(case(1, steps=3200))
+        b = sample_steps(case(2, steps=3200))
+        np.testing.assert_array_equal(a, b / 4)
+        self.assertEqual(len(a), 876)
+        self.assertEqual(a[-1], 3200)
+        self.assertTrue(all(t in a for t in range(0, 3201, 100)))
+        self.assertEqual(len(a[a <= 100]), 101)
+
+    def test_reference_factors(self):
+        f = factors(2)
+        self.assertEqual(f["mass"], 0.25)
+        self.assertEqual(f["impulse"], 0.5)
+        self.assertEqual(f["force"], 2)
+        self.assertEqual(f["mass_rate"], 1)
+
+    def test_extrapolation_polynomials(self):
+        self.assertAlmostEqual(LINEAR @ (2 + 3 * OFFSETS), 2)
+        self.assertAlmostEqual(QUADRATIC @ (2 + 3 * OFFSETS + 4 * OFFSETS**2), 2)
+
+    def test_independent_linear_field_volume_and_mass(self):
+        h = 0.5
+        y, x = np.mgrid[:64, :64]
+        x = (x + 0.5) * h
+        y = (y + 0.5) * h
+        rho = 1 + 0.001 * x
+        vel = np.array([0.01 + 0.0001 * x, -0.02 + 0.0002 * y])
+        center = np.array([16.0, 16.0])
+        r = 5
+        fluid = (x - center[0]) ** 2 + (y - center[1]) ** 2 > r * r
+        geo = ray_geometry(fluid, (0.0, 0.0), center, r)
+        ray, est = reconstruct_aperture(rho, vel, geo)
+        points = center + r * geo["normal"]
+        rr = 1 + 0.001 * points[..., 0]
+        uv = np.stack(
+            (0.01 + 0.0001 * points[..., 0], -0.02 + 0.0002 * points[..., 1]), axis=-1
+        )
+        un = np.sum(uv * geo["normal"], axis=-1)
+        exact = np.stack(
+            (np.sum(un * geo["ds"], axis=1), np.sum(rr * un * geo["ds"], axis=1)),
+            axis=-1,
+        )
+        np.testing.assert_allclose(est, np.broadcast_to(exact, est.shape), atol=1e-15)
+        self.assertGreater(np.max(abs(est[:, :, 0] - est[:, :, 1])), 1e-4)
+
+    def test_split_contour_stress(self):
+        y, x = np.mgrid[:48, :64]
+        y = (y + 0.5) * 0.5
+        rho = np.ones_like(y)
+        v = np.array([1e-4 * y * y, np.zeros_like(y)])
+        terms = contour_terms(rho, v, (0, 0), (5, 4, 25, 20), 0.0072)
+        np.testing.assert_allclose(terms[:2], 0, atol=1e-15)
+        np.testing.assert_allclose(terms[2], [-2 * 0.0072 * 1e-4 * 320, 0], atol=1e-15)
+
+    def test_recorder_snapshot_and_reconstructible_arrays(self):
+        h = 0.5
+        y, x = np.mgrid[:64, :64]
+        x = (x + 0.5) * h
+        y = (y + 0.5) * h
+        fluid = (x - 16) ** 2 + (y - 16) ** 2 > 25
+        fine_leaf = fluid.copy()
+        fine_leaf[:, :4] = False
+        root_leaf = np.zeros((32, 32), bool)
+        root_leaf[:, :2] = True
+
+        def equilibrium(ny, nx):
+            cu = CX[:, None, None] * 0.01
+            return np.broadcast_to(
+                W[:, None, None] * (1 + 3 * cu + 4.5 * cu * cu - 1.5 * 0.01**2),
+                (9, ny, nx),
+            ).copy()
+
+        ug = equilibrium(64, 64)
+        levels = [equilibrium(32, 32), ug.copy()]
+        c = dict(
+            scale=1,
+            center=[16, 16],
+            diameter=10,
+            u=0.01,
+            nu=0.0072,
+            contours=[[9, 9, 23, 23], [7, 7, 25, 25]],
+        )
+        data = {
+            name: [cv(ug, mask, (0, 0), r, c["nu"]) for r in c["contours"]]
+            for name, mask in [("uniform_adapter", fluid), ("amr_api", fine_leaf)]
+        }
+        with tempfile.TemporaryDirectory() as tmp:
+            rec = MeasurementRecorder(
+                c, tmp, fluid, [root_leaf, fine_leaf], (0, 0), CX, CY
+            )
+            rec.capture(0, ug, levels, data, {k: np.zeros(2) for k in data}, 0.0)
+            rec.records[-1]["raw_cumulative"] = np.zeros((2, 5, 4))
+            rec.flush()
+            a = np.load(Path(tmp) / "samples_s1.npz")
+            self.assertEqual(a["arc_ray_values"].shape, (1, 2, 3, 2, 32, 3))
+            np.testing.assert_allclose(
+                a["global_mass_reference"][0, 0], a["global_mass_reference"][0, 1]
+            )
+            snap = np.load(Path(tmp) / "fields_s1_tref0.npz")
+            np.testing.assert_allclose(
+                snap["uniform_velocity_reference"][0][fluid], 0.01, rtol=1e-7
+            )
+
+    def test_offline_mass_force_normalization(self):
+        import json
+
+        c = case(2, steps=3200)
+        s = 2
+        t = np.arange(0, 3201, 100)
+        n = len(t)
+        inv = np.zeros((n, 2, 2, 3))
+        inv[:, :, :, 0] = 100 * s * s
+        inv[:, :, :, 1] = -t[:, None, None] * 0.002 * s
+        raw = np.zeros((n, 2, 5, 4))
+        raw[:, :, 2, 0] = t[:, None] * 0.002 * s
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp)
+            (path / "samples_s2.json").write_text(
+                json.dumps(dict(case=c, sample_file="samples_s2.npz"))
+            )
+            np.savez(
+                path / "samples_s2.npz",
+                t_reference=t,
+                cv_inventory=inv,
+                cv_flux=np.zeros_like(inv),
+                raw_cumulative=raw,
+                arc_estimates=np.zeros((n, 2, 2, 2, 2)),
+                global_mass_reference=np.ones((n, 2)) * 100,
+                level_stats=np.broadcast_to([1, 1, 0, 0.03, 100], (n, 3, 5)),
+                velocity_error=np.zeros(n),
+            )
+            result = analyze(path)
+            self.assertEqual(len(result["windows"]), 32)
+            np.testing.assert_allclose(
+                result["windows"][0]["normalized_cv_residual"], 0, atol=1e-14
+            )
+            f0 = 0.5 * 0.03**2 * 24.084
+            np.testing.assert_allclose(
+                result["windows"][0]["raw_reaction_coefficient"], [[0.002 / f0, 0]] * 2
+            )
+
+    def test_nonuniform_trapezoidal_time_units(self):
+        t = np.array([0.0, 1.0, 2.0, 4.0, 8.0])
+        y = np.stack((3 * t + 2, 2 * t - 7), axis=-1)
+        np.testing.assert_allclose(integrate(t, y), [112, 8], atol=1e-15)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/validation/analyze_jet_cv_continuation.py

@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""Offline common-unit/sampling sensitivity comparison; never grants physics acceptance."""
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import numpy as np
+
+
+def integrate(t, y):
+    return np.sum(
+        np.diff(t).reshape((-1,) + (1,) * (y.ndim - 1)) * (y[1:] + y[:-1]) * 0.5, axis=0
+    )
+
+
+def analyze(directory):
+    meta_path = next(Path(directory).glob("samples_s*.json"))
+    meta = json.loads(meta_path.read_text())
+    c = meta["case"]
+    s = c["scale"]
+    sample_path = meta_path.parent / meta["sample_file"]
+    checksum_verified = False
+    if "sample_sha256" in meta:
+        if (
+            hashlib.sha256(sample_path.read_bytes()).hexdigest()
+            != meta["sample_sha256"]
+        ):
+            raise ValueError("Sample file hash mismatch")
+        checksum_verified = True
+    a = np.load(sample_path)
+    t = a["t_reference"]
+    if not len(t) or t[0] != 0 or np.any(np.diff(t) <= 0):
+        raise ValueError("Sample reference times must increase strictly from zero")
+    from jet_cv_measurements import sample_steps
+
+    complete_schedule = np.array_equal(t, sample_steps(c) / s**2)
+    records = []
+    # Input retains root units; convert before all budgets/comparisons.
+    inv = a["cv_inventory"] * np.array([1 / s**2, 1 / s, 1 / s])
+    flux = a["cv_flux"] * np.array([1.0, s, s])
+    raw = a["raw_cumulative"] * np.array([1 / s, 1 / s, 1.0, 1 / s**2])
+    raw[:, :, :, 2] /= (
+        s**2
+    )  # angular impulse reference = root/s^2 (not used in force tests)
+    q0 = 2 * np.pi / 12 * (c["diameter"] / 2) * c["jet_speed"]
+    f0 = 0.5 * (c["u"] * s) ** 2 * (c["diameter"] / s)
+    for lo in range(0, 3200, 100):
+        hi = lo + 100
+        ids = np.flatnonzero((t >= lo) & (t <= hi))
+        if len(ids) < 2 or t[ids[0]] != lo or t[ids[-1]] != hi:
+            continue
+        first, last = ids[0], ids[-1]
+        delta = inv[last] - inv[first]
+        ledger = raw[last] - raw[first]
+        cv_variants = []
+        for stride in (1, 2, 4):
+            selected = np.unique(np.r_[ids[::stride], ids[-1]])
+            total = integrate(t[selected], flux[selected])
+            m = delta[:, :, 0] + total[:, :, 0]
+            j = -delta[:, :, 1:] - total[:, :, 1:]
+            cv_variants.append(
+                np.concatenate(
+                    (
+                        (m - ledger[:, 2, 3, None])[:, :, None] / (100 * q0),
+                        (j - ledger[:, 2, None, :2]) / (100 * f0),
+                    ),
+                    axis=-1,
+                )
+            )
+        arc = (
+            integrate(t[ids], a["arc_estimates"][ids]) / 100
+        )  # Q/M rate conversion is unity
+        # Each estimator, aperture, Q/M; compare against signed prescribed Q (rho_ref=1 for M only).
+        arc_ratio = arc / np.array([q0, -q0])[None, None, :, None]
+        global_delta = (
+            a["global_mass_reference"][last] - a["global_mass_reference"][first]
+        )
+        records.append(
+            dict(
+                t_reference=[lo, hi],
+                startup=hi <= 800,
+                normalized_cv_residual=np.array(cv_variants).tolist(),
+                temporal_sensitivity_max=float(
+                    np.max(abs(np.array(cv_variants)[1:] - cv_variants[0]))
+                ),
+                arc_ratio_to_command_reference_density=arc_ratio.tolist(),
+                global_mass_residual_over_nominal=(
+                    (global_delta - ledger[:, 2, 3]) / (100 * q0)
+                ).tolist(),
+                raw_reaction_coefficient=(ledger[:, 2, :2] / (100 * f0)).tolist(),
+                raw_gross_mass_over_nominal=(
+                    (abs(ledger[:, 3, 3]) + abs(ledger[:, 4, 3])) / (100 * q0)
+                ).tolist(),
+                raw_wall_mass_over_nominal=(ledger[:, 0, 3] / (100 * q0)).tolist(),
+            )
+        )
+    return dict(
+        scale=s,
+        complete_schedule=bool(complete_schedule),
+        sample_checksum_verified=checksum_verified,
+        windows=records,
+        sampled_density_range=[
+            float(a["level_stats"][..., 0].min()),
+            float(a["level_stats"][..., 1].max()),
+        ],
+        sampled_max_speed_reference=float(a["level_stats"][..., 3].max()),
+        global_mass_initial_final=a["global_mass_reference"][[0, -1]].tolist(),
+        max_sampled_fine_velocity_error=float(a["velocity_error"].max()),
+        physics_validation_pass=False,
+    )
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("directories", nargs="+", type=Path)
+    p.add_argument("--output", type=Path, required=True)
+    a = p.parse_args()
+    results = [analyze(d) for d in a.directories]
+    payload = dict(
+        status="TRANSIENT_DIAGNOSTIC_COMPARISON",
+        results=results,
+        physics_validation_pass=False,
+    )
+    if sorted(x["scale"] for x in results) == [1, 2] and all(
+        x["complete_schedule"] for x in results
+    ):
+        ordered = sorted(results, key=lambda x: x["scale"])
+        summary = []
+        for startup in (True, False):
+            r = [
+                np.array(
+                    [
+                        w["normalized_cv_residual"][0]
+                        for w in item["windows"]
+                        if w["startup"] == startup
+                    ]
+                )
+                for item in ordered
+            ]
+            e = [
+                np.array(
+                    [
+                        w["temporal_sensitivity_max"]
+                        for w in item["windows"]
+                        if w["startup"] == startup
+                    ]
+                )
+                for item in ordered
+            ]
+            rms = [np.sqrt(np.mean(x * x, axis=(0, 1, 2))) for x in r]
+            unc = max(float(x.max()) for x in e)
+            summary.append(
+                dict(
+                    startup=startup,
+                    rms_normalized_residual_mass_Cx_Cy=[v.tolist() for v in rms],
+                    max_temporal_sensitivity=unc,
+                    reduction_exceeds_twice_temporal_proxy=(
+                        rms[0] - rms[1] > 2 * unc
+                    ).tolist(),
+                )
+            )
+        payload["refinement_trend"] = summary
+    a.output.write_text(json.dumps(payload, indent=2) + "\n")
+
+
+if __name__ == "__main__":
+    main()

tests/validation/jet_cv_measurements.py

@@ -0,0 +1,285 @@
+"""Measurement-only continuation diagnostics. Never changes populations or commands."""
+
+import hashlib
+import json
+from pathlib import Path
+import numpy as np
+
+OFFSETS = np.array([1.5, 2.5, 3.5])
+LINEAR = np.array([2.5, -1.5, 0.0])
+QUADRATIC = np.array([4.375, -5.25, 1.875])
+
+
+def sample_steps(c):
+    """Same reference times at both scales; dense startup retained."""
+    s2 = c["scale"] ** 2
+    reference = sorted(set(range(0, 101)) | set(range(100, 3201, 4)))
+    return np.array(reference, int) * s2
+
+
+def factors(scale):
+    """Lref=one s=1 root length, Tref=one s=1 root time, rho_ref=1."""
+    return dict(
+        length=1 / scale,
+        time=1 / scale**2,
+        velocity=scale,
+        mass=1 / scale**2,
+        momentum=1 / scale,
+        impulse=1 / scale,
+        force=scale,
+        volume_rate=1.0,
+        mass_rate=1.0,
+        pressure=scale**2,
+    )
+
+
+def primitives(g, cx, cy):
+    f = np.asarray(g, np.float64)
+    rho = f.sum(axis=0)
+    mom = np.stack([np.einsum("q,qyx->yx", v, f) for v in (cx, cy)])
+    return rho, mom / rho
+
+
+def contour_terms(rho, vel, origin, rectangle, nu, h=0.5):
+    """Independent split of total outward momentum flux into advective,p,viscous."""
+    x0, y0, x1, y1 = np.rint((np.array(rectangle) - np.tile(origin, 2)) / h).astype(int)
+    duy, dux = np.gradient(vel[0], h)
+    dvy, dvx = np.gradient(vel[1], h)
+    adv = rho * vel[:, None] * vel[None, :]
+    pressure = np.zeros_like(adv)
+    pressure[0, 0] = pressure[1, 1] = (rho - 1) / 3
+    viscous = -rho * nu * np.array([[2 * dux, duy + dvx], [duy + dvx, 2 * dvy]])
+    out = []
+    for field in (adv, pressure, viscous):
+        flux = np.zeros(2)
+        for i, sign in ((x0, -1), (x1, 1)):
+            flux += (
+                sign
+                * 0.5
+                * (field[:, 0, y0:y1, i - 1] + field[:, 0, y0:y1, i]).sum(axis=1)
+                * h
+            )
+        for j, sign in ((y0, -1), (y1, 1)):
+            flux += (
+                sign
+                * 0.5
+                * (field[:, 1, j - 1, x0:x1] + field[:, 1, j, x0:x1]).sum(axis=1)
+                * h
+            )
+        out.append(flux)
+    return np.array(out)
+
+
+def ray_geometry(fluid, origin, center, radius, h=0.5):
+    z, w = np.polynomial.legendre.leggauss(32)
+    theta = np.array([np.pi / 2, 3 * np.pi / 2])[:, None] + np.pi / 12 * z
+    normal = np.stack((np.cos(theta), np.sin(theta)), axis=-1)
+    positions = np.asarray(center) + normal[:, :, None, :] * (
+        radius + h * OFFSETS[None, None, :, None]
+    )
+    local = (positions - origin) / h - 0.5
+    ix = np.floor(local[..., 0]).astype(int)
+    iy = np.floor(local[..., 1]).astype(int)
+    a = local[..., 0] - ix
+    b = local[..., 1] - iy
+    if not (
+        np.all(ix >= 0)
+        and np.all(iy >= 0)
+        and np.all(ix + 1 < fluid.shape[1])
+        and np.all(iy + 1 < fluid.shape[0])
+    ):
+        raise ValueError("Ray interpolation leaves allocated grid")
+    for dy, dx in ((0, 0), (0, 1), (1, 0), (1, 1)):
+        if not fluid[iy + dy, ix + dx].all():
+            raise ValueError("Ray stencil must be active fluid")
+    return dict(
+        ix=ix,
+        iy=iy,
+        a=a,
+        b=b,
+        normal=normal,
+        ds=w * radius * np.pi / 12,
+        positions=positions,
+        origin=np.asarray(origin),
+        h=h,
+    )
+
+
+def reconstruct_aperture(rho, vel, geo):
+    field = np.concatenate((rho[None], vel), axis=0)
+    x, y, a, b = geo["ix"], geo["iy"], geo["a"], geo["b"]
+    ray = (
+        field[:, y, x] * (1 - a) * (1 - b)
+        + field[:, y, x + 1] * a * (1 - b)
+        + field[:, y + 1, x] * (1 - a) * b
+        + field[:, y + 1, x + 1] * a * b
+    )
+    estimates = []
+    for coeff in (LINEAR, QUADRATIC):
+        wall = np.einsum("cajk,k->caj", ray, coeff)
+        un = np.einsum("caj,ajc->aj", wall[1:], geo["normal"])
+        q = np.sum(un * geo["ds"], axis=1)
+        mass = np.sum(wall[0] * un * geo["ds"], axis=1)
+        estimates.append(np.stack((q, mass), axis=-1))
+    return ray, np.array(estimates)
+
+
+class MeasurementRecorder:
+    """Retain reconstructible per-sample terms and sparse full primitive snapshots."""
+
+    def __init__(self, c, output, fluid_u, leaves, origin_a, cx, cy):
+        self.c = c
+        self.output = Path(output)
+        self.fluid_u = fluid_u
+        self.leaves = leaves
+        self.origin_a = np.asarray(origin_a)
+        self.cx = cx
+        self.cy = cy
+        self.records = []
+        self.rays = [
+            ray_geometry(fluid_u, (0.0, 0.0), np.array(c["center"]), c["diameter"] / 2),
+            ray_geometry(leaves[1], origin_a, np.array(c["center"]), c["diameter"] / 2),
+        ]
+        self.snapshots = []
+
+    def capture(self, step, ug, levels, cv_data, nearest_flux, velocity_error):
+        c = self.c
+        s = c["scale"]
+        record = dict(
+            step=step,
+            t_reference=step / s**2,
+            t_star=step * c["u"] / c["diameter"],
+            velocity_error=velocity_error,
+        )
+        arrays = [ug, *levels]
+        masks = [self.fluid_u, *self.leaves]
+        hs = [0.5, 1.0, 0.5]
+        primitive = [primitives(g, self.cx, self.cy) for g in arrays]
+        stats = []
+        for (rho, vel), mask, h in zip(primitive, masks, hs):
+            if (
+                not np.isfinite(rho[mask]).all()
+                or rho[mask].min() <= 0
+                or not np.isfinite(vel[:, mask]).all()
+            ):
+                raise RuntimeError(
+                    "Invalid density/velocity on an active composite leaf"
+                )
+            stats.append(
+                [
+                    rho[mask].min(),
+                    rho[mask].max(),
+                    np.max(abs(rho[mask] - 1)),
+                    np.linalg.norm(vel[:, mask], axis=0).max() * s,
+                    rho[mask].sum() * h * h / s**2,
+                ]
+            )
+        record["level_stats"] = stats  # uniform, AMR root leaf, AMR fine leaf
+        record["global_mass_reference"] = [stats[0][-1], stats[1][-1] + stats[2][-1]]
+        record["cv_inventory"] = []
+        record["cv_flux"] = []
+        record["momentum_flux_terms"] = []
+        record["arc_ray_values"] = []
+        record["arc_estimates"] = []
+        record["nearest_mass_flux"] = [
+            nearest_flux[k] for k in ("uniform_adapter", "amr_api")
+        ]
+        for idx, (name, p, origin) in enumerate(
+            [
+                ("uniform_adapter", primitive[0], (0.0, 0.0)),
+                ("amr_api", primitive[2], self.origin_a),
+            ]
+        ):
+            rho, vel = p
+            record["cv_inventory"].append([x[0] for x in cv_data[name]])
+            record["cv_flux"].append([x[1] for x in cv_data[name]])
+            terms = np.array(
+                [contour_terms(rho, vel, origin, r, c["nu"]) for r in c["contours"]]
+            )
+            np.testing.assert_allclose(
+                terms.sum(axis=1),
+                np.array(record["cv_flux"][-1])[:, 1:],
+                atol=2e-12,
+                rtol=2e-10,
+            )
+            record["momentum_flux_terms"].append(terms)
+            ray, estimates = reconstruct_aperture(rho, vel, self.rays[idx])
+            record["arc_ray_values"].append(ray)
+            record["arc_estimates"].append(estimates)
+        self.records.append(record)
+        if step // s**2 in (0, 100, 800, 1600, 3200) and step % s**2 == 0:
+            fields = {}
+            for name, (rho, vel), mask, h, origin in zip(
+                ("uniform", "amr_root", "amr_fine"),
+                primitive,
+                masks,
+                hs,
+                ((0.0, 0.0), (0.0, 0.0), self.origin_a),
+            ):
+                fields[name + "_rho"] = np.where(mask, rho, np.nan).astype(np.float32)
+                fields[name + "_velocity_reference"] = np.where(
+                    mask[None], vel * s, np.nan
+                ).astype(np.float32)
+                fields[name + "_origin_reference"] = np.asarray(origin) / s
+                fields[name + "_h_reference"] = np.array(h / s)
+            file = self.output / f"fields_s{s}_tref{step//s**2}.npz"
+            np.savez_compressed(file, **fields)
+            self.snapshots.append(
+                dict(
+                    file=file.name,
+                    sha256=hashlib.sha256(file.read_bytes()).hexdigest(),
+                    bytes=file.stat().st_size,
+                    t_reference=step / s**2,
+                )
+            )
+
+    def flush(self):
+        path = self.output / f"samples_s{self.c['scale']}.npz"
+        arrays = {k: np.array([r[k] for r in self.records]) for k in self.records[0]}
+        for i, g in enumerate(self.rays):
+            for k in ("normal", "ds", "positions"):
+                arrays[f"ray_{i}_{k}"] = g[k]
+        arrays["ray_offsets_fine_cells"] = OFFSETS
+        arrays["linear_coefficients"] = LINEAR
+        arrays["quadratic_coefficients"] = QUADRATIC
+        np.savez_compressed(path, **arrays)
+        metadata = dict(
+            schema=1,
+            case=self.c,
+            reference_conversion=factors(self.c["scale"]),
+            sample_file=path.name,
+            sample_sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
+            sample_count=len(self.records),
+            snapshots=self.snapshots,
+            level_stats_columns=[
+                "rho_min",
+                "rho_max",
+                "max_abs_rho_minus_1",
+                "max_speed_reference",
+                "mass_reference",
+            ],
+            raw_partition_order=["wall", "aperture", "total", "blowing", "suction"],
+            raw_columns=["Jx", "Jy", "angular_impulse", "mass_into_fluid"],
+            backend_order=["uniform_adapter", "amr_api"],
+            level_order=["uniform", "amr_root_leaf", "amr_fine_leaf"],
+            momentum_term_order=["advection", "pressure", "negative_viscous_traction"],
+            arc_estimate_axes=[
+                "sample",
+                "backend",
+                "linear_or_quadratic",
+                "blowing_or_suction",
+                "Q_or_M",
+            ],
+            arc_ray_axes=[
+                "sample",
+                "backend",
+                "rho_ux_uy",
+                "aperture",
+                "gauss_point",
+                "normal_offset",
+            ],
+            warning="Extrapolation and nearest-node values are estimator-dependent, not exact wall values. Fields/ray arrays use root-lattice units except explicitly named reference arrays.",
+        )
+        (self.output / f"samples_s{self.c['scale']}.json").write_text(
+            json.dumps(metadata, indent=2) + "\n"
+        )

tests/validation/run_jet_circle_cv.py

@@ -202,7 +202,7 @@ def verify(repo):
     return manifest
 
 
-def run_case(c, ctx, output, provenance):
+def run_case(c, ctx, output, provenance, continuation=False):
     from CelerisLab.body.geometry.circle import CircleGeometry
     from CelerisLab.body.objects import SimObject
     from CelerisLab.body.jets import Jet, Sector, Radial
@@ -334,9 +334,26 @@ def run_case(c, ctx, output, provenance):
             "amr_api": aperture_probe(fluid_a, origin_a, center, radius),
         }
 
-        def capture():
+        recorder = None
+        if continuation:
+            from jet_cv_measurements import MeasurementRecorder, sample_steps
+
+            recorder = MeasurementRecorder(
+                c, output, fluid_u, amr._leaf, origin_a, CX, CY
+            )
+            scheduled = set(sample_steps(c).tolist())
+        else:
+            scheduled = set(range(c["steps"] + 1))
+
+        cumulative_raw = {
+            name: {k: np.zeros(4) for k in masks}
+            for name in ("uniform_adapter", "amr_api")
+        }
+
+        def capture(step):
             ug = uniform.download()
-            ag = amr.download()[1]
+            levels = amr.download()
+            ag = levels[1]
             data = {}
             aperture = {}
             for name, g, fluid, origin in [
@@ -356,9 +373,18 @@ def run_case(c, ctx, output, provenance):
                 np.linalg.norm((av - uv)[:, fluid_a])
                 / max(np.linalg.norm(uv[:, fluid_a]), 1e-30)
             )
+            if recorder is not None:
+                recorder.capture(step, ug, levels, data, aperture, error)
+                recorder.records[-1]["raw_cumulative"] = np.array(
+                    [
+                        [cumulative_raw[name][k].copy() for k in masks]
+                        for name in ("uniform_adapter", "amr_api")
+                    ]
+                )
             return data, error, aperture
 
-        previous, _, previous_aperture = capture()
+        previous, _, previous_aperture = capture(0)
+        previous_sample_step = 0
         initial = previous
         integrated = {name: np.zeros((2, 3)) for name in previous}
         integrated_aperture = {name: np.zeros(2) for name in previous}
@@ -380,9 +406,9 @@ def run_case(c, ctx, output, provenance):
             flush=True,
         )
         for step in range(1, c["steps"] + 1):
-            if time.monotonic() - start > 1800:
+            if time.monotonic() - start > c.get("post_ready_budget_seconds", 1800):
                 raise RuntimeError(
-                    "1800-second post-READY pilot budget exceeded; partial report retained"
+                    "Declared post-READY budget exceeded; partial report retained"
                 )
             samples.clear()
             amr.step()
@@ -393,25 +419,38 @@ def run_case(c, ctx, output, provenance):
             ar = np.empty((count, 4), np.float64)
             with ctx.activate():
                 cuda.memcpy_dtoh(ar, b.interval_integral)
+            if not np.isfinite(ar).all() or not np.isfinite(ur).all():
+                raise RuntimeError("Nonfinite raw exchange in completed base interval")
             obs = amr.read_raw_exchange(0)
             if obs["time_start"] != step - 1 or obs["time_end"] != step:
                 raise RuntimeError("Raw interval time mismatch")
             np.testing.assert_allclose(
                 ar.sum(axis=0)[:2], obs["total"]["impulse"], atol=1e-13, rtol=1e-13
             )
-            current, error, current_aperture = capture()
             for name, rr in [("uniform_adapter", ur), ("amr_api", ar)]:
                 for k, mask in masks.items():
-                    accumulated[name][k] += rr[mask].sum(axis=0)
-                integrated_aperture[name] += 0.5 * (
-                    previous_aperture[name] + current_aperture[name]
+                    subtotal = rr[mask].sum(axis=0)
+                    accumulated[name][k] += subtotal
+                    cumulative_raw[name][k] += subtotal
+            if step not in scheduled:
+                continue
+            current, error, current_aperture = capture(step)
+            sample_dt = step - previous_sample_step
+            for name in current:
+                integrated_aperture[name] += (
+                    0.5 * sample_dt * (previous_aperture[name] + current_aperture[name])
                 )
-                integrated[name] += 0.5 * (
-                    np.array([x[1] for x in previous[name]])
-                    + np.array([x[1] for x in current[name]])
+                integrated[name] += (
+                    0.5
+                    * sample_dt
+                    * (
+                        np.array([x[1] for x in previous[name]])
+                        + np.array([x[1] for x in current[name]])
+                    )
                 )
             previous = current
             previous_aperture = current_aperture
+            previous_sample_step = step
             if step % c["window"] == 0 or step == c["steps"]:
                 end0 = rows[-1]["step_end"] if rows else 0
                 duration = step - end0
@@ -469,6 +508,8 @@ def run_case(c, ctx, output, provenance):
                         v.fill(0)
                 rows.append(row)
                 initial = current
+                if recorder is not None:
+                    recorder.flush()
                 payload = dict(
                     case=c,
                     source_revision=PIN,
@@ -506,19 +547,40 @@ def main():
         "--steps", type=int, default=800, help="Base s=1 steps; scales as s squared"
     )
     p.add_argument("--allow-re400", action="store_true")
+    p.add_argument(
+        "--continuation",
+        action="store_true",
+        help="Fixed measurement campaign: Re100, scale1/2, 3200 reference steps",
+    )
     p.add_argument("--plan", action="store_true")
     a = p.parse_args()
     if a.steps <= 0 or a.steps > 3200:
         p.error("steps must be 1..3200")
     if a.re == 400 and not a.allow_re400:
         p.error("Re400 requires explicit conditional-campaign opt-in")
+    if a.continuation:
+        if a.scale not in (1, 2) or a.re != 100 or a.steps != 3200:
+            p.error("--continuation requires --re 100 --scale 1/2 --steps 3200")
     c = case(a.scale, a.re, a.steps)
+    if a.continuation:
+        from jet_cv_measurements import sample_steps, factors
+
+        c.update(
+            measurement_schema=1,
+            sample_count=len(sample_steps(c)),
+            startup_sample_dt_reference=1,
+            later_sample_dt_reference=4,
+            dense_startup_end_reference=100,
+            reference_conversion=factors(a.scale),
+            post_ready_budget_seconds=1200 if a.scale == 1 else 5400,
+            runtime_estimate_seconds=[210, 500] if a.scale == 1 else [900, 3200],
+        )
     plan = dict(
         case=c,
         required_revision=PIN,
         required_free_device_mib=1024,
         estimated_peak_device_mib=128 * a.scale**2,
-        estimated_output_mib=2,
+        estimated_output_mib=64 if a.continuation else 2,
         max_steps=c["steps"],
         gpu_launch_owner="root; this runner does not select or reserve a GPU",
     )
@@ -550,7 +612,7 @@ def main():
             free, total = cuda.mem_get_info()
         if free < max(1024, 2 * plan["estimated_peak_device_mib"]) * 1024**2:
             raise RuntimeError("Insufficient free device memory")
-        run_case(c, ctx, a.output, provenance)
+        run_case(c, ctx, a.output, provenance, continuation=a.continuation)
     print(
         json.dumps(
             dict(status="COMPLETE_DIAGNOSTIC_ONLY", physics_validation_pass=False)