From 724afb607cb81d6bc64f3b2df5845a33b4430f64 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Oct 2026 12:29:21 +0000 Subject: [PATCH] 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 --- docs/validation/refined_dfg_timing.md | 5 ++- tests/unit/test_time_refined_dfg.py | 15 +++++++ tests/validation/time_refined_dfg.py | 59 +++++++++++++++++++++++++-- 3 files changed, 74 insertions(+), 5 deletions(-) diff --git a/docs/validation/refined_dfg_timing.md b/docs/validation/refined_dfg_timing.md index 214a5fe..a052e3b 100644 --- a/docs/validation/refined_dfg_timing.md +++ b/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. diff --git a/tests/unit/test_time_refined_dfg.py b/tests/unit/test_time_refined_dfg.py index 25183b8..126b152 100644 --- a/tests/unit/test_time_refined_dfg.py +++ b/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 diff --git a/tests/validation/time_refined_dfg.py b/tests/validation/time_refined_dfg.py index 6d89c33..7d1a055 100644 --- a/tests/validation/time_refined_dfg.py +++ b/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][:, fluid]).all(): + if not np.isfinite(values[0].reshape(values[0].shape[0], -1)[:, fluid.ravel()]).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}")