test: fix shear-wave review findings (runner finiteness, source pin, contract)

- run_case: finite_row accepts list diagnostics; sample() shared by sample
  and READY paths; verify_source pins shear_wave.py path+SHA256 for
  --flow shear_wave (default TG flow unchanged, does not import it).
- assess_shear_wave.py: fixed N128 four-case contract and 4-JSON CLI,
  reusing assess_projected_pairs provenance/build/workload/time checks plus
  shear metadata; missing/duplicate/mixed/extra records -> INVALID_SET.
- tests: mocked refined runner sample (list diagnostics, NaN, TG unchanged),
  contract pass + reproduced false-pass mutations, divergence check uses
  analytic E.k=0 and a derived h^2 truncation + roundoff bound with
  h-halving scaling instead of a fixed 1e-12.

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

tests/validation/taylor_green/test_shear_wave.py

@@ -79,8 +79,21 @@ class ConventionTests(unittest.TestCase):
         dy = (f(x, y + h, t) - f(x, y - h, t)) / (2 * h)
         lap = (f(x + h, y, t) + f(x - h, y, t) + f(x, y + h, t) + f(x, y - h, t) - 4 * z) / h**2
         np.testing.assert_allclose(dt, NU * lap, atol=1e-9)
-        np.testing.assert_allclose(z[0] * dx + z[1] * dy, 0.0, atol=1e-12)
-        np.testing.assert_allclose(dx[0] + dy[1], 0.0, atol=1e-12)
+        # Analytic identities: div u = A cos(theta) (E.k) and (u.grad)u = A sin(theta) (E.k) u/A.
+        k = sw.wavevector(N)
+        self.assertLess(abs(float(sw.E @ k)), 1e-17)
+        # Central differences do not cancel exactly: each derivative carries
+        # -(k_j h)^2/6 relative truncation, so div_h = -A cos(theta) h^2/6 sum_j E_j k_j^3
+        # + O(h^4) (nonzero for an oblique k). Bound it with that leading term plus roundoff.
+        trunc = U * h * h / 6 * float(np.abs(sw.E * k**3).sum()) * 1.01
+        # Argument rounding: eps*max|x|/h*U*|k| ~ 2.2e-16*21.6/1e-3*0.03*0.44 ~ 6e-14.
+        roundoff = 1e-13
+        self.assertLess(np.max(np.abs(dx[0] + dy[1])), trunc + roundoff)
+        self.assertLess(np.max(np.abs(z[0] * dx + z[1] * dy)), U * trunc + roundoff)
+        # Halving h reduces the divergence truncation about fourfold.
+        hh = h / 2
+        div2 = ((f(x + hh, y, t) - f(x - hh, y, t))[0] + (f(x, y + hh, t) - f(x, y - hh, t))[1]) / (2 * hh)
+        np.testing.assert_allclose(div2 * 4, dx[0] + dy[1], rtol=1e-2, atol=6 * roundoff)
         self.assertTrue(np.all(sw.exact(x, y, t, N, NU)[2] == 0))
 
 
@@ -256,6 +269,124 @@ class AssessTests(unittest.TestCase):
         self.assertEqual(sw.assess(base, fine, [refined])["status"], "FAIL")
 
 
+class FakeRun:
+    def __init__(self, arrays):
+        self.arrays = arrays
+
+    def download(self):
+        return self.arrays if len(self.arrays) > 1 else self.arrays[0]
+
+
+class RunnerSampleTests(unittest.TestCase):
+    def test_refined_shear_sample_with_list_diagnostics(self):
+        masks = composite_masks(N)
+        arrays = [sw.initial(m.shape, level, N, NU) for level, m in enumerate(masks)]
+        row = run_case.sample(FakeRun(arrays), masks, 0, N, NU, None, sw.metrics, 2)
+        self.assertIsInstance(row["spurious_pressure_scaled_by_level"], list)
+        self.assertTrue(run_case.finite_row(row))
+        bad = [a.copy() for a in arrays]
+        bad[1][0, masks[1]] = np.nan
+        with self.assertRaises(FloatingPointError):
+            run_case.sample(FakeRun(bad), masks, 0, N, NU, None, sw.metrics, 2, "READY")
+        self.assertFalse(run_case.finite_row(dict(a=1.0, b=[0.0, float("inf")])))
+
+    def test_taylor_green_sample_unchanged(self):
+        masks = composite_masks(N)
+        arrays = [oracle.initial(m.shape, level, N, NU) for level, m in enumerate(masks)]
+        row = run_case.sample(FakeRun(arrays), masks, 0, N, NU, None, oracle.metrics, 2)
+        self.assertEqual(row, oracle.metrics(arrays, masks, 0, N, NU, None))
+        uniform = [oracle.initial((N, N), 0, N, NU)]
+        row = run_case.sample(FakeRun(uniform), [np.ones((N, N), bool)], 0, N, NU, None,
+                              oracle.metrics, 1)
+        self.assertEqual(row["leaf_cells"], N * N)
+
+
+def receipt(n, depth, explosion="constant"):
+    """Synthetic receipt satisfying every contract field of the fixed four-case set."""
+    import assess_projected_pairs as tg
+    steps = run_case.sample_steps(n)[0]
+    rows = rows_for(n)
+    for row, step in zip(rows, steps):
+        row.update(step=step, tstar=step * U / n)
+    cells = [n * n] if depth == 1 else [8704, 30720]
+    hops = [16 * n * 2**level for level in range(depth)]
+    builds = []
+    for level in range(depth):
+        size = n * 2**level
+        builds.append(dict(source_sha256=tg.BUILD_SOURCES[size], arch="sm_70",
+                           flags=["--use_fast_math"], n_objects=0, ptx_sha256="a" * 64,
+                           config=dict(nx=10, ny=10, nz=1, lattice_model="D2Q9", collision="SRT",
+                                       store_precision="FP32", streaming="double_buffer",
+                                       ddf_shifting=True, les_enabled=False, velocity=0.0,
+                                       viscosity=U * size / 100, rho=1.0,
+                                       omega=1 / (.5 + 3 * U * size / 100),
+                                       threads_per_block=256)))
+    checkout = "/x/CelerisLab"
+    return dict(
+        n=n, depth=depth, re_L=100, explosion=explosion, U=U, mach=U * math.sqrt(3),
+        viscosity=U * n / 100, tau_by_level=[0.5 + 3 * U * n / 100 * 2**l for l in range(depth)],
+        source=dict(revision="r", dirty=False, source_tree_sha256=run_case.SOURCE_SHA256,
+                    harness_sha256={"x.py": "h"}, build_capture_sha256=run_case.BUILD_CAPTURE_SHA256,
+                    shear_wave_sha256=run_case.SHEAR_WAVE_SHA256),
+        runtime_identity=dict(device=dict(name="V100", pci_bus_id="0", capability=[7, 0],
+                                          driver=1, runtime=[12, 0]),
+                              nvcc_path="/nvcc", nvcc_version="v", python="3", numpy="1",
+                              pycuda="2"),
+        runtime_source=dict(checkout=checkout, module=checkout + "/src/CelerisLab/__init__.py",
+                            compiler=checkout + "/src/CelerisLab/cuda/compiler_v2.py",
+                            kernel_root=checkout + "/src/CelerisLab/lbm/kernels"),
+        cuda_visible_devices="0", numerical_revision=run_case.NUMERICAL_REVISION,
+        reference_harness=run_case.REFERENCE_HARNESS, numerics=dict(tg.NUMERICS),
+        initialization=sw.INITIALIZATION, limits=dict(sw.LIMITS),
+        early_pressure=dict(enabled=False, rows=[], sampled_max_pressure_l2=None),
+        refinements=[] if depth == 1 else rectangles(n, depth),
+        workload=dict(active_cells_by_level=cells, level_hops=hops, root_intervals=16 * n,
+                      active_collision_updates_by_level=[c * h for c, h in zip(cells, hops)],
+                      active_collision_updates=sum(c * h for c, h in zip(cells, hops))),
+        actual_build_sidecars=builds, rows=rows, completed=True,
+        flow="shear_wave", mode=list(sw.MODE), polarization=sw.E.tolist(),
+        angle_deg=sw.ANGLE_DEG, phase=sw.PHI, eps=sw.eps(n))
+
+
+class ContractTests(unittest.TestCase):
+    def setUp(self):
+        import assess_shear_wave
+        self.assess = assess_shear_wave.assess
+        self.records = [receipt(128, 1), receipt(256, 1), receipt(128, 2),
+                        receipt(128, 2, "projected_linear")]
+
+    def test_complete_contract_passes(self):
+        result = self.assess(self.records, "r")
+        self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v])
+        self.assertEqual(result["status"], "PASS")
+
+    def test_reproduced_false_passes_are_rejected(self):
+        a = self.assess
+        self.assertEqual(a(self.records[:2], "r")["status"], "INVALID_SET")  # empty refined
+        dup = [*self.records[:3], receipt(128, 2)]
+        self.assertEqual(a(dup, "r")["status"], "INVALID_SET")  # duplicate constant-only
+        for mutate in (
+            lambda r: r.pop("source"),
+            lambda r: r.update(flow="taylor_green"),
+            lambda r: r.update(mode=[1, 2]),
+            lambda r: r.update(re_L=1000),
+            lambda r: r.update(refinements=[[(0, 0, 8, 8)]]),
+            lambda r: r["actual_build_sidecars"][0].update(arch="sm_80"),
+            lambda r: r["rows"][0].update(step=1, tstar=U / 128),  # wrong initial time
+            lambda r: r["source"].update(shear_wave_sha256="0" * 64),
+        ):
+            records = [dict(r, rows=[dict(x) for x in r["rows"]],
+                            source=dict(r["source"]),
+                            actual_build_sidecars=[dict(b) for b in r["actual_build_sidecars"]])
+                       for r in self.records]
+            mutate(records[2])
+            self.assertEqual(a(records, "r")["status"], "INVALID_SET")
+        mixed = [dict(r) for r in self.records]
+        mixed[1] = dict(mixed[1], source=dict(mixed[1]["source"], revision="other"))
+        self.assertEqual(a(mixed, "r")["status"], "INVALID_SET")
+        self.assertEqual(a(self.records, "wrong-revision")["status"], "INVALID_SET")
+
+
 class DefaultPathTests(unittest.TestCase):
     def test_frozen_oracle_is_byte_identical(self):
         digest = hashlib.sha256((HERE / "oracle.py").read_bytes()).hexdigest()