test: physical health and stricter parity in timing harness

Untimed physical-leaf/fluid health after warm-up and after each timed
window using the existing runner limits (finite macros, rho in [0.5,1.5],
max Mach <= 0.2); refined masks as in run_refined_dfg. Parity comparator
rejects unequal level counts and any infinity; matching NaN stays allowed
only as a storage mask. Regression tests for both. No numerical gate or
production change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/validation/refined_dfg_timing.md

@@ -21,7 +21,10 @@ physics-gate or production change; no force or accuracy result is produced.
   every 100 steps in per-step loops).
 - Final populations and NaN masks must be bitwise identical to the first A repeat (A/B/C) or the
   first E repeat (E/E'); refined and uniform grids are never compared. Leaf/fluid populations
-  must be finite. All parity/health checks are untimed.
+  must be finite. Physical health (finite leaf/fluid macros, rho in [0.5, 1.5], max Mach <= 0.2,
+  the existing runner limits) is checked after warm-up and after each timed window. Matching NaN
+  is accepted only as a storage mask; any infinity or level-count mismatch fails parity. All
+  parity/health checks are untimed.
 - Cases run sequentially (uniform, then refined) in one process with startup serialized;
   variants are interleaved within each repeat. A-B and B-C are measured comparative
   differences under GPU variability, not exact causal decompositions.

tests/unit/test_time_refined_dfg.py

@@ -42,6 +42,21 @@ class TimingHarnessContract(unittest.TestCase):
         self.assertIn("bits", t.same_state([a], [b]))
         c = a.copy(); c[0, 0] = np.nan
         self.assertIn("NaN", t.same_state([a], [c]))
+        self.assertIn("level count", t.same_state([a], [a, a]))
+        inf = a.copy(); inf[1, 0] = np.inf
+        self.assertIn("infinite", t.same_state([inf], [inf.copy()]))  # matching inf still fails
+        self.assertIn("infinite", t.same_state([a], [inf]))
+
+    def test_physical_health_limits(self):
+        ok = (np.array([1., 1.01]), np.array([.01, .02]), np.array([0., .01]))
+        health = t.physical_health([ok])
+        self.assertAlmostEqual(health["rho_max"], 1.01)
+        for bad in ((np.array([1., np.nan]),) + ok[1:],
+                    (np.array([.4, 1.]),) + ok[1:],
+                    (ok[0], np.array([.2, 0.]), ok[2]),
+                    (np.array([]), np.array([]), np.array([]))):
+            with self.assertRaises(RuntimeError):
+                t.physical_health([bad])
 
     def test_summary_reports_differences_without_claims(self):
         rows = [dict(variant=v, seconds=s) for v, s in

tests/validation/time_refined_dfg.py

@@ -53,12 +53,25 @@ def schedule(n_base, warmup_base):
                 tstar_window=n_base * r.SPEED / 40.0)
 
 
+# Existing run_refined_dfg safety limits (rho in [.5, 1.5], sampled max Mach <= .2).
+RHO_LIMITS, MAX_MACH = (0.5, 1.5), 0.2
+
+
 def same_state(reference, values):
-    """Bitwise populations and identical NaN masks; returns a short description or None."""
+    """Bitwise populations and identical NaN masks; returns a short description or None.
+
+    Matching NaN is accepted only as a storage mask (inactive/ghost/solid slots); physical
+    leaf health is checked separately. Any infinity in either state is a mismatch.
+    """
+    reference, values = list(reference), list(values)
+    if len(reference) != len(values):
+        return f"level count {len(reference)} != {len(values)}"
     for level, (a, b) in enumerate(zip(reference, values)):
         a, b = np.asarray(a), np.asarray(b)
         if a.shape != b.shape:
             return f"level {level}: shape {a.shape} != {b.shape}"
+        if np.isinf(a).any() or np.isinf(b).any():
+            return f"level {level}: infinite population"
         if not np.array_equal(np.isnan(a), np.isnan(b)):
             return f"level {level}: NaN mask differs"
         finite = np.isfinite(a)
@@ -67,6 +80,31 @@ def same_state(reference, values):
     return None
 
 
+def physical_health(fields):
+    """fields: (rho, ux, uy) arrays already restricted to physical leaves/fluid."""
+    if not fields or any(not np.isfinite(v).all() or v.size == 0 for f in fields for v in f):
+        raise RuntimeError("nonfinite or empty physical macroscopic field")
+    rho_min = float(min(f[0].min() for f in fields))
+    rho_max = float(max(f[0].max() for f in fields))
+    max_mach = float(max(np.sqrt(3 * (f[1] ** 2 + f[2] ** 2)).max() for f in fields))
+    if not (RHO_LIMITS[0] <= rho_min <= rho_max <= RHO_LIMITS[1] and max_mach <= MAX_MACH):
+        raise RuntimeError(f"health limits violated: rho [{rho_min}, {rho_max}], Mach {max_mach}")
+    return dict(rho_min=rho_min, rho_max=rho_max, max_mach=max_mach)
+
+
+def refined_health(backend, masks):
+    arrays = backend.download()  # physical FP64 populations; non-leaves NaN by contract
+    if any(not np.isfinite(g[:, m]).all() for g, m in zip(arrays, masks)):
+        raise RuntimeError("nonfinite physical leaf population")
+    return physical_health([tuple(v[m] for v in r.macros(g)) for g, m in zip(arrays, masks)])
+
+
+def uniform_health(sim, fluid):
+    m = sim.get_macroscopic()
+    return physical_health([tuple(np.asarray(m[k], np.float64).reshape(fluid.shape)[fluid]
+                                  for k in ("rho", "ux", "uy"))])
+
+
 def check_deadline(deadline):
     if time.monotonic() >= deadline:
         raise TimeoutError("timing harness exceeded --max-wall-seconds")
@@ -111,6 +149,9 @@ def time_refined(a, h, sched):
     from contextlib import ExitStack
     from CelerisLab.cuda.context import CudaContext
     from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+    from CelerisLab.lbm.channel import channel_masks
+    from CelerisLab.lbm.refinement import ACTIVE
+    from CelerisLab.lbm.descriptors import SOLID
 
     warm, n = sched["refined"]
     rows, reference = [], {}
@@ -120,6 +161,8 @@ def time_refined(a, h, sched):
         backend = stack.enter_context(FixedPatchStepper(
             cfg, r.BOUNDS, context=context, coarse_boundary="channel", base_block=r.BASE_BLOCK,
             fine_block=r.FINE_BLOCK, fine_objects=(body,), observe_raw_exchange=True))
+        masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered,  # as run_refined_dfg
+                 (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)]
         fresh = (r.initial_state(pair.coarse_shape, (0.0, 0.0), 1.0, root=True),
                  r.initial_state(pair.fine_shape, pair.fine_origin, 0.5))
         print(json.dumps(dict(status="READY", case="refined")), flush=True)
@@ -128,11 +171,13 @@ def time_refined(a, h, sched):
                 check_deadline(a.deadline)
                 backend.initialize(*fresh)  # resets step count; same fresh state every time
                 refined_loop(backend, variant, warm, a.deadline)
+                warm_health = refined_health(backend, masks)  # untimed
                 sync_refined(backend)
                 start = time.perf_counter()
                 refined_loop(backend, variant, n, a.deadline)
                 sync_refined(backend)
                 seconds = time.perf_counter() - start
+                end_health = refined_health(backend, masks)  # untimed
                 values = [np.array(v, copy=True) for v in backend._download_raw().values]
                 if not all(np.isfinite(v[:, m]).all() for v, m in zip(values, backend._state)):
                     raise RuntimeError(f"refined {variant} repeat {repeat}: nonfinite leaf populations")
@@ -143,6 +188,7 @@ def time_refined(a, h, sched):
                     mismatch = same_state(reference["A"], values)
                 rows.append(dict(variant=variant, repeat=repeat, seconds=seconds,
                                  base_steps=n, step_count=backend.step_count,
+                                 health_after_warmup=warm_health, health_after_window=end_health,
                                  parity_vs_A=mismatch or "identical", **VARIANTS[variant]))
                 if mismatch:
                     raise RuntimeError(f"refined {variant} repeat {repeat}: {mismatch}")
@@ -155,7 +201,7 @@ def time_uniform(a, h, sched):
 
     warm, n = sched["uniform"]
     rows, reference = [], {}
-    cfgdict, bodydict, _ = r.config(h, 80, "MRT")
+    cfgdict, bodydict, params = r.config(h, 80, "MRT")
     with TemporaryDirectory() as tmp:
         tmp = Path(tmp)
         (tmp / "lbm.json").write_text(json.dumps(cfgdict))
@@ -163,6 +209,9 @@ def time_uniform(a, h, sched):
         with Simulation(str(tmp / "lbm.json"), str(tmp / "body.json"), device_id=0) as sim:
             sim.initialize()
             sim.snapshot()  # fresh state; restore keeps clock and streaming parity
+            fluid = (sim.get_flags() & FLUID) != 0
+            ny, nx = params["ny"], params["nx"]
+            fluid = fluid.reshape(ny, nx)
             print(json.dumps(dict(status="READY", case="uniform")), flush=True)
             for repeat in range(a.repeats):
                 for variant in ("E", "E'"):
@@ -171,14 +220,15 @@ def time_uniform(a, h, sched):
                     if sim.stepper.step_count % 2:
                         raise RuntimeError("uniform restore parity changed between variants")
                     uniform_loop(sim, variant, warm, a.deadline)
+                    warm_health = uniform_health(sim, fluid)  # untimed
                     sync_uniform(sim)
                     start = time.perf_counter()
                     uniform_loop(sim, variant, n, a.deadline)
                     sync_uniform(sim)
                     seconds = time.perf_counter() - start
+                    end_health = uniform_health(sim, fluid)  # untimed
                     values = [sim.get_ddf()]
-                    fluid = (sim.get_flags().reshape(values[0].shape[1:]) & FLUID) != 0
+                    if not np.isfinite(values[0].reshape(values[0].shape[0], -1)[:, fluid.ravel()]).all():
-                    if not np.isfinite(values[0][:, fluid]).all():
                         raise RuntimeError(f"uniform {variant} repeat {repeat}: nonfinite fluid populations")
                     mismatch = None
                     if repeat == 0 and variant == "E":
@@ -187,6 +237,7 @@ def time_uniform(a, h, sched):
                         mismatch = same_state(reference["E"], values)
                     rows.append(dict(variant=variant, repeat=repeat, seconds=seconds,
                                      uniform_steps=n, parity_vs_E=mismatch or "identical",
+                                     health_after_warmup=warm_health, health_after_window=end_health,
                                      **VARIANTS[variant]))
                     if mismatch:
                         raise RuntimeError(f"uniform {variant} repeat {repeat}: {mismatch}")