test: add bounded dense-root Jet CV temporal diagnostics

tests/validation/analyze_jet_cv_continuation.py

@@ -32,7 +32,7 @@ def analyze(directory):
     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
+    from jet_cv_measurements import sample_steps, old_sample_steps
 
     complete_schedule = np.array_equal(t, sample_steps(c) / s**2)
     records = []
@@ -45,7 +45,15 @@ def analyze(directory):
     )  # 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):
+    dense_root = c.get("dense_root", False)
+    old_times = old_sample_steps(c) / s**2
+    ray_indices = (
+        a["ray_sample_indices"] if "ray_sample_indices" in a else np.arange(len(t))
+    )
+    ray_times = t[ray_indices]
+    if dense_root and complete_schedule and not np.array_equal(ray_times, old_times):
+        raise ValueError("Sparse ray records do not match the declared old cadence")
+    for lo in range(0, c.get("steps", 3200 * s * s) // (s * s), 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:
@@ -54,8 +62,14 @@ def analyze(directory):
         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]])
+        selections = [np.unique(np.r_[ids[::stride], ids[-1]]) for stride in (1, 2, 4)]
+        if dense_root:
+            selections.append(ids[np.isin(t[ids], old_times)])
+        for selected in selections:
+            if t[selected[0]] != lo or t[selected[-1]] != hi:
+                raise ValueError(
+                    "Every quadrature selection must retain both window endpoints"
+                )
             total = integrate(t[selected], flux[selected])
             m = delta[:, :, 0] + total[:, :, 0]
             j = -delta[:, :, 1:] - total[:, :, 1:]
@@ -68,16 +82,61 @@ def analyze(directory):
                     axis=-1,
                 )
             )
+        ray_ids = np.flatnonzero((ray_times >= lo) & (ray_times <= hi))
         arc = (
-            integrate(t[ids], a["arc_estimates"][ids]) / 100
+            integrate(ray_times[ray_ids], a["arc_estimates"][ray_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]
         )
+        dense_details = {}
+        if dense_root:
+            absolute = (
+                a["raw_absolute_aperture_mass_cumulative"][last]
+                - a["raw_absolute_aperture_mass_cumulative"][first]
+            ) / s**2
+            stock_raw = np.diff(a["global_mass_reference"][ids], axis=0) - np.diff(
+                raw[ids, :, 2, 3], axis=0
+            )
+            partition = ledger[:, 2] - ledger[:, 0] - ledger[:, 1]
+            dense_details = dict(
+                quadrature_names=[
+                    "root_stride1",
+                    "root_stride2",
+                    "root_stride4",
+                    "exact_old_cadence",
+                ],
+                root_dt_reference=1 / s**2,
+                old_cadence_change_from_dense=(
+                    cv_variants[3] - cv_variants[0]
+                ).tolist(),
+                root_temporal_sensitivity_max=float(
+                    np.max(abs(np.array(cv_variants)[1:3] - cv_variants[0]))
+                ),
+                raw_absolute_aperture_net_interval_mass_reference=absolute.tolist(),
+                prescribed_unsigned_nominal_mass_reference=float(2 * q0 * 100),
+                legacy_raw_gross_field_definition="raw_gross_mass_over_nominal retains abs(window net blowing)+abs(window net suction), divided by ONE nominal aperture throughput; it is not root-interval absolute throughput.",
+                raw_absolute_to_prescribed_unsigned_nominal=(
+                    absolute / (2 * q0 * 100)
+                ).tolist(),
+                global_stock_raw_balance_max_abs_root_interval_reference=np.max(
+                    abs(stock_raw), axis=0
+                ).tolist(),
+                global_stock_raw_balance_sum_reference=stock_raw.sum(axis=0).tolist(),
+                raw_total_minus_wall_aperture_reference=partition.tolist(),
+                raw_aperture_minus_blowing_suction_reference=(
+                    ledger[:, 1] - ledger[:, 3] - ledger[:, 4]
+                ).tolist(),
+                measured_ray_absolute_volume_mass_reference=integrate(
+                    ray_times[ray_ids], a["arc_absolute_estimates"][ray_ids]
+                ).tolist(),
+                measured_ray_cadence="Sparse old cadence; estimator-dependent, not dense-root or exact wall throughput",
+            )
         records.append(
             dict(
+                **dense_details,
                 t_reference=[lo, hi],
                 startup=hi <= 800,
                 normalized_cv_residual=np.array(cv_variants).tolist(),
@@ -97,6 +156,12 @@ def analyze(directory):
         )
     return dict(
         scale=s,
+        dense_root=dense_root,
+        previous_temporal_failure_retained=(
+            "Prior continuation stride2 changes exceeded baseline residual in all48 audited comparisons; this diagnostic does not retroactively change those gates."
+            if dense_root
+            else None
+        ),
         complete_schedule=bool(complete_schedule),
         sample_checksum_verified=checksum_verified,
         windows=records,
@@ -122,8 +187,10 @@ def main():
         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
+    if (
+        not any(x.get("dense_root", False) for x in results)
+        and 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 = []

tests/validation/jet_cv_measurements.py

@@ -10,13 +10,63 @@ LINEAR = np.array([2.5, -1.5, 0.0])
 QUADRATIC = np.array([4.375, -5.25, 1.875])
 
 
+EVIDENCE_LIMIT = 32 * 1024**2
+
+
+def install_evidence(replacements, limit=None):
+    """Preflight retained bytes before replacing files; cap failure preserves old pair.
+
+    Staged compression bytes are temporary disk overhead, not retained evidence.
+    This guards budget failures, not a transactional crash-recovery filesystem.
+    """
+    temporary = set(replacements.values())
+    try:
+        directory = next(iter(replacements)).parent
+        retained = sum(
+            p.stat().st_size
+            for p in directory.iterdir()
+            if p.is_file() and p not in temporary
+        )
+        projected = retained - sum(p.stat().st_size for p in replacements if p.exists())
+        projected += sum(p.stat().st_size for p in temporary)
+        if limit is not None and projected > limit:
+            raise RuntimeError(
+                "Dense-root 32 MiB retained-evidence budget would be exceeded; last verified files preserved"
+            )
+        for target, staged in replacements.items():
+            staged.replace(target)
+    finally:
+        for p in temporary:
+            if p.exists():
+                p.unlink()
+
+
+def write_evidence_text(path, text, *, dense_root=False):
+    staged = path.with_name("." + path.name + ".pending")
+    staged.write_text(text)
+    install_evidence({path: staged}, EVIDENCE_LIMIT if dense_root else None)
+
+
 def sample_steps(c):
     """Same reference times at both scales; dense startup retained."""
+    if c.get("dense_root", False):
+        return np.arange(c["steps"] + 1, dtype=int)
+    return old_sample_steps(c)
+
+
+def old_sample_steps(c):
+    """Exact previous physical cadence, clipped to this campaign's endpoint."""
     s2 = c["scale"] ** 2
-    reference = sorted(set(range(0, 101)) | set(range(100, 3201, 4)))
+    end = c.get("steps", 3200 * s2) // s2
+    reference = sorted(set(range(0, min(100, end) + 1)) | set(range(100, end + 1, 4)))
     return np.array(reference, int) * s2
 
 
+def absolute_aperture_interval_mass(rows, blowing, suction):
+    """Sum absolute aperture NET masses for one root interval, not per-link flux."""
+    return float(abs(rows[blowing, 3].sum()) + abs(rows[suction, 3].sum()))
+
+
 def factors(scale):
     """Lref=one s=1 root length, Tref=one s=1 root time, rho_ref=1."""
     return dict(
@@ -105,7 +155,7 @@ def ray_geometry(fluid, origin, center, radius, h=0.5):
     )
 
 
-def reconstruct_aperture(rho, vel, geo):
+def reconstruct_aperture(rho, vel, geo, *, return_absolute=False):
     field = np.concatenate((rho[None], vel), axis=0)
     x, y, a, b = geo["ix"], geo["iy"], geo["a"], geo["b"]
     ray = (
@@ -115,12 +165,24 @@ def reconstruct_aperture(rho, vel, geo):
         + field[:, y + 1, x + 1] * a * b
     )
     estimates = []
+    absolute = []
     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))
+        absolute.append(
+            np.stack(
+                (
+                    np.sum(abs(un) * geo["ds"], axis=1),
+                    np.sum(abs(wall[0] * un) * geo["ds"], axis=1),
+                ),
+                axis=-1,
+            )
+        )
+    if return_absolute:
+        return ray, np.array(estimates), np.array(absolute)
     return ray, np.array(estimates)
 
 
@@ -136,6 +198,8 @@ class MeasurementRecorder:
         self.cx = cx
         self.cy = cy
         self.records = []
+        self.sparse_rays = []
+        self.ray_steps = set(old_sample_steps(c).tolist())
         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),
@@ -179,8 +243,16 @@ class MeasurementRecorder:
         record["cv_inventory"] = []
         record["cv_flux"] = []
         record["momentum_flux_terms"] = []
-        record["arc_ray_values"] = []
-        record["arc_estimates"] = []
+        save_ray = not c.get("dense_root", False) or step in self.ray_steps
+        ray_record = dict(
+            sample_index=len(self.records),
+            arc_ray_values=[],
+            arc_estimates=[],
+            arc_absolute_estimates=[],
+        )
+        if not c.get("dense_root", False):
+            record["arc_ray_values"] = []
+            record["arc_estimates"] = []
         record["nearest_mass_flux"] = [
             nearest_flux[k] for k in ("uniform_adapter", "amr_api")
         ]
@@ -203,9 +275,18 @@ class MeasurementRecorder:
                 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)
+            if save_ray:
+                ray, estimates, absolute = reconstruct_aperture(
+                    rho, vel, self.rays[idx], return_absolute=True
+                )
+                ray_record["arc_ray_values"].append(ray)
+                ray_record["arc_estimates"].append(estimates)
+                ray_record["arc_absolute_estimates"].append(absolute)
+                if not c.get("dense_root", False):
+                    record["arc_ray_values"].append(ray)
+                    record["arc_estimates"].append(estimates)
+        if c.get("dense_root", False) and save_ray:
+            self.sparse_rays.append(ray_record)
         self.records.append(record)
         if step // s**2 in (0, 100, 800, 1600, 3200) and step % s**2 == 0:
             fields = {}
@@ -223,7 +304,11 @@ class MeasurementRecorder:
                 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)
+            staged = file.with_name("." + file.name + ".pending.npz")
+            np.savez_compressed(staged, **fields)
+            install_evidence(
+                {file: staged}, EVIDENCE_LIMIT if c.get("dense_root", False) else None
+            )
             self.snapshots.append(
                 dict(
                     file=file.name,
@@ -236,19 +321,32 @@ class MeasurementRecorder:
     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]}
+        if self.c.get("dense_root", False):
+            arrays["ray_sample_indices"] = np.array(
+                [r["sample_index"] for r in self.sparse_rays], dtype=int
+            )
+            for key in ("arc_ray_values", "arc_estimates", "arc_absolute_estimates"):
+                arrays[key] = np.array([r[key] for r in self.sparse_rays])
         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)
+        staged = path.with_name("." + path.name + ".pending.npz")
+        np.savez_compressed(staged, **arrays)
         metadata = dict(
-            schema=1,
+            schema=2 if self.c.get("dense_root", False) else 1,
             case=self.c,
+            dense_root=bool(self.c.get("dense_root", False)),
+            ray_sample_index_key=(
+                "ray_sample_indices" if self.c.get("dense_root", False) else None
+            ),
+            raw_absolute_throughput_definition="Cumulative sum of abs(blowing NET dm)+abs(suction NET dm) per root interval; unresolved within-aperture/substep reversals; not abs(window net).",
+            absolute_throughput_definition="Quadrature integral of abs(reconstructed u.n), abs(reconstructed rho*u.n); not abs(signed integral), not prescribed nominal, not exact wall truth.",
             reference_conversion=factors(self.c["scale"]),
             sample_file=path.name,
-            sample_sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
+            sample_sha256=hashlib.sha256(staged.read_bytes()).hexdigest(),
             sample_count=len(self.records),
             snapshots=self.snapshots,
             level_stats_columns=[
@@ -280,6 +378,10 @@ class MeasurementRecorder:
             ],
             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"
+        meta_path = self.output / f"samples_s{self.c['scale']}.json"
+        meta_staged = meta_path.with_name("." + meta_path.name + ".pending")
+        meta_staged.write_text(json.dumps(metadata, indent=2) + "\n")
+        install_evidence(
+            {path: staged, meta_path: meta_staged},
+            EVIDENCE_LIMIT if self.c.get("dense_root", False) else None,
         )

tests/validation/run_jet_circle_cv.py

@@ -350,6 +350,10 @@ def run_case(c, ctx, output, provenance, continuation=False):
             for name in ("uniform_adapter", "amr_api")
         }
 
+        from jet_cv_measurements import absolute_aperture_interval_mass
+
+        cumulative_absolute_aperture_mass = np.zeros(2)
+
         def capture(step):
             ug = uniform.download()
             levels = amr.download()
@@ -381,6 +385,10 @@ def run_case(c, ctx, output, provenance, continuation=False):
                         for name in ("uniform_adapter", "amr_api")
                     ]
                 )
+                if c.get("dense_root", False):
+                    recorder.records[-1][
+                        "raw_absolute_aperture_mass_cumulative"
+                    ] = cumulative_absolute_aperture_mass.copy()
             return data, error, aperture
 
         previous, _, previous_aperture = capture(0)
@@ -427,7 +435,17 @@ def run_case(c, ctx, output, provenance, continuation=False):
             np.testing.assert_allclose(
                 ar.sum(axis=0)[:2], obs["total"]["impulse"], atol=1e-13, rtol=1e-13
             )
-            for name, rr in [("uniform_adapter", ur), ("amr_api", ar)]:
+            np.testing.assert_allclose(
+                ar[:, 3].sum(), obs["total"]["mass_into_fluid"], atol=1e-13, rtol=1e-13
+            )
+            for backend_index, (name, rr) in enumerate(
+                [("uniform_adapter", ur), ("amr_api", ar)]
+            ):
+                cumulative_absolute_aperture_mass[
+                    backend_index
+                ] += absolute_aperture_interval_mass(
+                    rr, masks["blowing"], masks["suction"]
+                )
                 for k, mask in masks.items():
                     subtotal = rr[mask].sum(axis=0)
                     accumulated[name][k] += subtotal
@@ -519,8 +537,12 @@ def run_case(c, ctx, output, provenance, continuation=False):
                     physics_validation_pass=False,
                     precision="FP32 evolution, FP64 host diagnostics",
                 )
-                (output / f"re{c['re_d']}_s{c['scale']}.json").write_text(
-                    json.dumps(payload, indent=2) + "\n"
+                from jet_cv_measurements import write_evidence_text
+
+                write_evidence_text(
+                    output / f"re{c['re_d']}_s{c['scale']}.json",
+                    json.dumps(payload, indent=2) + "\n",
+                    dense_root=c.get("dense_root", False),
                 )
                 print(
                     json.dumps(
@@ -552,12 +574,22 @@ def main():
         action="store_true",
         help="Fixed measurement campaign: Re100, scale1/2, 3200 reference steps",
     )
+    p.add_argument(
+        "--dense-root",
+        action="store_true",
+        help="Bounded ref0..800 scalar diagnostics at every root interval; sparse rays/fields",
+    )
     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.dense_root:
+        if a.continuation or a.scale not in (1, 2) or a.re != 100 or a.steps != 800:
+            p.error(
+                "--dense-root requires --re 100 --scale 1/2 --steps 800, without --continuation"
+            )
     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")
@@ -575,12 +607,28 @@ def main():
             post_ready_budget_seconds=1200 if a.scale == 1 else 5400,
             runtime_estimate_seconds=[210, 500] if a.scale == 1 else [900, 3200],
         )
+    if a.dense_root:
+        from jet_cv_measurements import sample_steps, old_sample_steps, factors
+
+        c.update(
+            dense_root=True,
+            measurement_schema=2,
+            reference_conversion=factors(a.scale),
+            post_ready_budget_seconds=1200 if a.scale == 1 else 5400,
+            runtime_estimate_seconds=[60, 240] if a.scale == 1 else [600, 2400],
+        )
+        c.update(
+            sample_count=len(sample_steps(c)),
+            sparse_ray_sample_count=len(old_sample_steps(c)),
+            scalar_sample_dt_reference=1 / a.scale**2,
+            raw_interval_dt_reference=1 / a.scale**2,
+        )
     plan = dict(
         case=c,
         required_revision=PIN,
         required_free_device_mib=1024,
         estimated_peak_device_mib=128 * a.scale**2,
-        estimated_output_mib=64 if a.continuation else 2,
+        estimated_output_mib=32 if a.dense_root else (64 if a.continuation else 2),
         max_steps=c["steps"],
         gpu_launch_owner="root; this runner does not select or reserve a GPU",
     )
@@ -612,7 +660,9 @@ 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, continuation=a.continuation)
+        run_case(
+            c, ctx, a.output, provenance, continuation=a.continuation or a.dense_root
+        )
     print(
         json.dumps(
             dict(status="COMPLETE_DIAGNOSTIC_ONLY", physics_validation_pass=False)