test: add bounded square shear orientation diagnostic

tests/validation/taylor_green/run_case.py

@@ -27,6 +27,8 @@ TWIN_SHEAR = "twin_shear_n128"  # depth-2 twin: nested L0->L1 boxes without L2
 SHEAR_TWIN_SHA256 = "1cbf37f18e89aea577581631c4565f51e68d22a3739b698816ef64acbf2b2bb4"
 WIDE_X_SHEAR = "wide_x_shear_n128"  # one projected x-gap-width contrast
 SHEAR_WIDE_X_SHA256 = "13d16ba355117e2d074edccd9dc5d694c2fc13c673a0abdfca6401f6ed917733"
+AXIS_SQUARE = "axis_square_shear_n128"  # short four-face orientation diagnostic
+SHEAR_AXIS_SQUARE_SHA256 = "0baffa6eb99c268ce998d90300c993d45bb3b7b27c524b636874d75808bd0a8c"
 POSTG_SHEAR = "postg_shear_n128"  # initialization-only three-case counterfactual
 SHEAR_POSTG_SHA256 = "9c909fa013c092490ee21532e3125b2a78a4c5102c7a1bdc0e76986d3f0ef903"
 SHEAR_PROBES_SHA256 = "e2e6490d281cc275f448dc6d6ea7b9b4407cc74a51e9256b86c08c8626b89b76"  # signed interface-probe mechanism diagnostic
@@ -125,6 +127,13 @@ def verify_source(repo, revision, flow="taylor_green", probes=False):
                 or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_PROBES_SHA256):
             raise RuntimeError("Wrong shear_probes module checkout or content")
         result["shear_probes_sha256"] = SHEAR_PROBES_SHA256
+    if flow == "shear_axis_square":
+        import shear_axis_square
+        module = harness / "shear_axis_square.py"
+        if (Path(shear_axis_square.__file__).resolve() != module
+                or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_AXIS_SQUARE_SHA256):
+            raise RuntimeError("Wrong shear_axis_square module checkout or content")
+        result["shear_axis_square_sha256"] = SHEAR_AXIS_SQUARE_SHA256
     if flow == "shear_postg":
         import shear_postg
         module = harness / "shear_postg.py"
@@ -153,7 +162,7 @@ def verify_source(repo, revision, flow="taylor_green", probes=False):
                 or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_NESTED_SHA256):
             raise RuntimeError("Wrong shear_nested module checkout or content")
         result["shear_nested_sha256"] = SHEAR_NESTED_SHA256
-    if flow in ("shear_wave", "shear_nested", "shear_twin", "shear_wide_x", "shear_postg"):
+    if flow in ("shear_wave", "shear_nested", "shear_twin", "shear_wide_x", "shear_postg", "shear_axis_square"):
         import shear_wave
         module = harness / "shear_wave.py"
         if (Path(shear_wave.__file__).resolve() != module
@@ -228,16 +237,18 @@ def main():
     p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
     p.add_argument("--diagnostic",
                    choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR, NESTED_SHEAR, TWIN_SHEAR,
-                            WIDE_X_SHEAR, POSTG_SHEAR],
+                            WIDE_X_SHEAR, POSTG_SHEAR, AXIS_SQUARE],
                    help="explicit four-case N256/fine512 same-placement diagnostic "
                         "(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave) or the "
                         "fixed three-level/two-region shear preset (nested_shear_n128) "
                         "or its depth-2 twin (twin_shear_n128), or the single projected "
                         "x-gap-width contrast (wide_x_shear_n128), or three-case POST-G "
-                        "initialization counterfactual (postg_shear_n128)")
+                        "initialization counterfactual (postg_shear_n128), or short square "
+                        "orientation diagnostic (axis_square_shear_n128)")
     p.add_argument("--probe-trace", action="store_true",
                    help="signed interface-probe trace (nested U128/U512/projected d3, "
-                        "twin projected d2, wide-x projected d3, POST-G three-case set); "
+                        "twin projected d2, wide-x projected d3, POST-G three-case set, "
+                        "axis-square four-case set); "
                         "mechanism diagnostic, no gates")
     p.add_argument("--final-snapshot", action="store_true",
                    help="nested_shear_n128 projected depth-3 only: final macro snapshot")
@@ -245,10 +256,11 @@ def main():
                    help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
     p.add_argument("--flow", choices=["taylor_green", "shear_wave"], default="taylor_green")
     args = p.parse_args()
+    axis_square = args.diagnostic == AXIS_SQUARE
     nested = args.diagnostic == NESTED_SHEAR
     wide_x = args.diagnostic == WIDE_X_SHEAR
     postg = args.diagnostic == POSTG_SHEAR
-    if (wide_x or postg) and not args.probe_trace:
+    if (wide_x or postg or axis_square) and not args.probe_trace:
         p.error(f"{args.diagnostic} requires --probe-trace")
     if args.flow == "shear_wave" and (args.re != 100 or (args.depth == 3 and not (nested or wide_x or postg))
                                       or args.early_pressure):
@@ -259,9 +271,13 @@ def main():
                    (WIDE_X_SHEAR, 128, 3, "projected_linear"),
                    (POSTG_SHEAR, 128, 1, "constant"), (POSTG_SHEAR, 512, 1, "constant"),
                    (POSTG_SHEAR, 128, 3, "projected_linear")}
+    if axis_square:
+        import shear_axis_square
+        probe_cases.update((AXIS_SQUARE, n, depth, scheme)
+                           for _, n, depth, scheme in shear_axis_square.CASES)
     if args.probe_trace and (args.final_snapshot or (
             args.diagnostic, args.n, args.depth, args.explosion) not in probe_cases):
-        p.error("--probe-trace requires a fixed nested, twin, wide-x or POST-G case, "
+        p.error("--probe-trace requires a fixed nested, twin, wide-x, POST-G or axis-square case, "
                 "without --final-snapshot")
     if args.final_snapshot and not (nested and args.depth == 3
                                     and args.explosion == "projected_linear"):
@@ -272,7 +288,12 @@ def main():
     if args.explosion != "constant" and args.depth != 2 and not (nested or wide_x or postg):
         p.error("projected_linear requires depth 2")
     twin = args.diagnostic == TWIN_SHEAR
-    if nested or twin or wide_x or postg:
+    if axis_square:
+        preset = shear_axis_square
+        if args.flow != "shear_wave" or (args.re, args.n, args.depth, args.explosion) not in preset.CASES:
+            p.error("axis_square_shear_n128 requires one of its four Re100 shear cases")
+        initial, metrics, limits, initialization = preset.initial, preset.metrics, {}, preset.INITIALIZATION
+    elif nested or twin or wide_x or postg:
         import shear_nested
         import shear_twin
         import shear_wide_x
@@ -304,7 +325,7 @@ def main():
         p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
     repo = args.repo.resolve()
     sys.path[:0] = [str(repo / "src"), str(repo)]
-    source_flow = ("shear_postg" if postg else "shear_wide_x" if wide_x else "shear_nested" if nested
+    source_flow = ("shear_axis_square" if axis_square else "shear_postg" if postg else "shear_wide_x" if wide_x else "shear_nested" if nested
                    else "shear_twin" if twin else args.flow)
     source = verify_source(repo, args.expected_revision, source_flow, probes=args.probe_trace)
     runtime = runtime_source(repo)
@@ -322,7 +343,7 @@ def main():
     cfg = config(args.n, args.re)
     if args.depth == 1:
         refinements = []
-    elif nested or twin or wide_x or postg:
+    elif nested or twin or wide_x or postg or axis_square:
         refinements = [list(level) for level in preset.REFINEMENTS]
     else:
         refinements = rectangles(args.n, args.depth)
@@ -330,6 +351,8 @@ def main():
     np.testing.assert_array_equal(leaf_partition(h), np.ones((args.n, args.n)))
     masks = [k == ACTIVE for k in h.kinds]
     main_steps, early_steps = sample_steps(args.n, args.early_pressure)
+    if axis_square:
+        main_steps = [0, preset.case_steps(args.n)[-1]]
     report = dict(
         schema=1,
         source=source,
@@ -357,7 +380,7 @@ def main():
         cuda_visible_devices=selected,
         device_id=0,
         limits=limits,
-        workload=workload(masks, round(TIMES[-1] * args.n / U)),
+        workload=workload(masks, main_steps[-1]),
         rows=[],
         early_pressure=dict(enabled=args.early_pressure, rows=[],
                             sampled_max_pressure_l2=None),
@@ -365,14 +388,19 @@ def main():
     )
     if args.flow != "taylor_green":
         import shear_wave
-        report.update(flow=args.flow, mode=list(shear_wave.MODE),
-                      polarization=shear_wave.E.tolist(), angle_deg=shear_wave.ANGLE_DEG,
-                      phase=shear_wave.PHI, eps=shear_wave.eps(args.n))
+        wave = preset if axis_square else shear_wave
+        report.update(flow=args.flow, mode=list(wave.MODE),
+                      polarization=wave.E.tolist(), angle_deg=wave.ANGLE_DEG,
+                      phase=wave.PHI, eps=wave.eps(args.n))
     if args.diagnostic:
         report["diagnostic"] = args.diagnostic
     if postg:
         report.update(initialization_id=preset.INITIALIZATION_ID, qualification_eligible=False,
                       screening_scope="descriptive scalar screens only; no qualification assessor")
+    if axis_square:
+        report.update(initialization_id=preset.INITIALIZATION_ID, qualification_eligible=False,
+                      reference_harness=None, historical_baseline_revision=None,
+                      screening_scope="no numerical gates; short orientation diagnostic")
     args.output.parent.mkdir(parents=True, exist_ok=True)
     def save():
         temporary = args.output.with_suffix(args.output.suffix + ".tmp")
@@ -410,6 +438,8 @@ def main():
             probe_steps = set()
             if args.probe_trace:
                 import shear_probes
+                if axis_square:
+                    shear_probes = preset
                 probe_root = dict(zip(shear_probes.case_steps(args.n), shear_probes.STEPS))
                 probe_steps = set(probe_root)
                 probes = shear_probes.select(masks, args.n)
@@ -491,7 +521,7 @@ def main():
     report.update(
         completed=True,
         seconds=time.monotonic() - started,
-        screening_pass=all(
+        screening_pass=None if axis_square else all(
             row[key] <= limit for row in report["rows"] for key, limit in limits.items()
         ),
     )
@@ -502,6 +532,8 @@ def main():
         ),
         flush=True,
     )
+    if axis_square:
+        return int("error" in report["probe_trace"])
     return 0 if report["screening_pass"] else 1
 
 

tests/validation/taylor_green/shear_axis_square.py

@@ -0,0 +1,136 @@
+"""Short axis-shear orientation diagnostic; no physical acceptance gates."""
+
+import json
+import numpy as np
+
+import shear_wave as sw
+from shear_probes import sample, validate_row, TRACE_MAX_BYTES
+from oracle import CX, CY, W, U, centers
+
+ROOT_N = 128
+MODE, E, PHI, ANGLE_DEG = (2, 0), np.array([0., 1.]), np.pi / 4, 0.
+REFINEMENTS = [[(32, 32, 96, 96)]]
+CASES = ((100, 128, 1, "constant"), (100, 256, 1, "constant"),
+         (100, 128, 2, "constant"), (100, 128, 2, "projected_linear"))
+STEPS = tuple(range(9))
+INITIALIZATION_ID = "postg_first_gradient_v1"
+INITIALIZATION = "physical FP64 axis-shear POST G=feq-(tau-1)A; first-gradient low-Mach CE"
+GROUPS = dict(inner_left=(32, 64), inner_right=(95, 63),
+              inner_bottom=(64, 32), inner_top=(63, 95),
+              outer_left=(31, 64), outer_right=(96, 63),
+              outer_bottom=(64, 31), outer_top=(63, 96))
+
+
+def eps(n):
+    return (4 * np.pi / n)**2
+
+
+def exact(x, t, length, viscosity):
+    """Transverse velocity and physical x derivative in the supplied length units."""
+    k = 4 * np.pi / length
+    amplitude = U * np.exp(-viscosity * k * k * t)
+    return amplitude * np.sin(k * x + PHI), amplitude * k * np.cos(k * x + PHI)
+
+
+def initial(shape, level, length, viscosity):
+    x, _ = centers(shape, level)
+    uy, gradient = exact(x, 0, length, viscosity)
+    tau = .5 + 3 * viscosity * 2**level
+    a = (3 * W * CX * CY)[:, None, None] * gradient * 2.**-level
+    return sw.equilibrium(np.ones_like(uy), np.zeros_like(uy), uy) - (tau - 1) * a
+
+
+def metrics(arrays, masks, t, length, viscosity, initial_mass=None):
+    rho, ux, uy, x, _, w, _ = sw._leaves(arrays, masks)
+    if not np.isclose(w.sum(), length**2, rtol=0, atol=1e-10):
+        raise ValueError("leaf partition does not cover domain once")
+    ey, _ = exact(x, t, length, viscosity)
+    p = (rho - 1) / 3
+    p -= w @ p / w.sum()  # one domain gauge, never the probe mean
+    mass = float(w @ rho)
+    return dict(tstar=t * U / length, mass=mass,
+                mass_relative=0. if initial_mass is None else abs(mass-initial_mass)/initial_mass,
+                velocity_l2=float(np.sqrt(w @ (ux**2 + (uy-ey)**2) / (w @ ey**2))),
+                spurious_pressure_rms=float(np.sqrt(w @ p**2 / w.sum())),
+                rho_min=float(rho.min()), rho_max=float(rho.max()),
+                max_speed=float(np.hypot(ux, uy).max()), leaf_cells=len(rho))
+
+
+def case_steps(n):
+    return [r * (n // ROOT_N) for r in STEPS]
+
+
+def select(masks, n=ROOT_N):
+    """Exact native cells filling eight unit-area groups; no interpolated samples."""
+    if (n, len(masks)) not in ((128, 1), (256, 1), (128, 2)):
+        raise ValueError("unsupported square probe layout")
+    probes = []
+    for name, (x, y) in GROUPS.items():
+        level = int(len(masks) == 2 and name.startswith("inner"))
+        factor = (n // ROOT_N) * 2**level
+        for dy in range(factor):
+            for dx in range(factor):
+                i, j = factor*x + dx, factor*y + dy
+                if not masks[level][j, i]:
+                    raise ValueError("square probe is not ACTIVE")
+                probes.append(dict(group=name, level=level, i=i, j=j,
+                                   center=[(i+.5)/factor, (j+.5)/factor],
+                                   width=1/factor, area=1/factor**2, parity=[i % 2, j % 2]))
+    if len({(p["level"], p["i"], p["j"]) for p in probes}) != len(probes):
+        raise ValueError("duplicate square probes")
+    return probes
+
+
+def new_trace(probes, case_viscosity, case):
+    return dict(schema=1, kind="axis-square orientation diagnostic; no pass gates", case=case,
+                time_unit="root step (N=128)", sample_phase="postcollision G at root synchronization",
+                root_n=ROOT_N, viscosity_root=U*ROOT_N/100, case_viscosity=case_viscosity,
+                mode=list(MODE), phase=PHI, amplitude=U, fields=["rho", "ux", "uy"],
+                probes=probes, steps=[], values=[])
+
+
+def group_moments(trace):
+    """Signed unit-area means; normalize by each group's mean actual-center derivative.
+
+    Momentum is rho*u. Raw density is ungauged. Per-cell parity stays in the trace.
+    U128 samples one center, whereas U256/refined-inner use four different centers.
+    """
+    out = []
+    for step, row in zip(trace["steps"], trace["values"]):
+        values, groups = np.asarray(row).reshape(-1, 3), {}
+        for name in GROUPS:
+            indices = [i for i, p in enumerate(trace["probes"]) if p["group"] == name]
+            probes = [trace["probes"][i] for i in indices]
+            weights = np.array([p["area"] for p in probes])
+            if weights.sum() != 1:
+                raise ValueError("probe group does not have unit area")
+            rho, ux, uy = values[indices].T
+            ey, gradient = exact(np.array([p["center"][0] for p in probes]), step,
+                                 ROOT_N, trace["viscosity_root"])
+            drho, djx, djy = (float(weights @ v) for v in (rho-1, rho*ux, rho*uy-ey))
+            derivative = float(weights @ gradient)
+            groups[name] = dict(drho=drho, djx=djx, djy=djy, mean_duy_dx=derivative,
+                                djx_over_duy_dx=djx/derivative, djy_over_duy_dx=djy/derivative)
+        out.append(dict(root_step=step, groups=groups))
+    return out
+
+
+def validate(trace):
+    if trace["steps"] != list(STEPS) or len(trace["values"]) != len(STEPS):
+        raise ValueError("square trace schedule mismatch")
+    _, n, depth, scheme = trace["case"]
+    if (100, n, depth, scheme) not in CASES or len(trace["probes"]) != (
+            20 if depth == 2 else 8*(n//ROOT_N)**2):
+        raise ValueError("square trace case/count mismatch")
+    for row in trace["values"]:
+        validate_row(row, len(trace["probes"]))
+    trace["group_moments"] = group_moments(trace)
+    size = 0
+    while True:
+        final = len(json.dumps(dict(trace, bytes=size), allow_nan=False).encode())
+        if final == size:
+            break
+        size = final
+    if size > TRACE_MAX_BYTES:
+        raise ValueError("square trace exceeds byte cap")
+    return size

tests/validation/taylor_green/test_axis_square.py

@@ -0,0 +1,172 @@
+"""Fixed square geometry, native sampling, POST phase and runner admission; CPU only."""
+import hashlib
+import io
+import json
+import sys
+import tempfile
+import unittest
+from copy import deepcopy
+from pathlib import Path
+from unittest.mock import patch
+from contextlib import redirect_stdout
+
+import numpy as np
+import run_case as rc
+import shear_axis_square as axis
+import shear_wave as sw
+import test_shear_wave as legacy
+from oracle import CX, CY, W, U, centers
+from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition
+from CelerisLab.lbm.refinement import ACTIVE
+
+REPO = Path(__file__).resolve().parents[3]
+
+
+def masks(n, depth):
+    h = build_hierarchy(rc.config(n, 100), [] if depth == 1 else axis.REFINEMENTS)
+    np.testing.assert_array_equal(leaf_partition(h), np.ones((n, n)))
+    return [k == ACTIVE for k in h.kinds]
+
+
+class AxisSquareTests(unittest.TestCase):
+    def test_geometry_native_groups_and_work(self):
+        work = 0
+        for _, n, depth, _ in axis.CASES:
+            m = masks(n, depth)
+            p = axis.select(m, n)
+            self.assertEqual(len(p), 20 if depth == 2 else 8*(n//128)**2)
+            self.assertEqual(len({(v['level'], v['i'], v['j']) for v in p}), len(p))
+            roots = [(32, 64), (95, 63), (64, 32), (63, 95),
+                     (31, 64), (96, 63), (64, 31), (63, 96)]
+            expected = []
+            for group, (x, y) in enumerate(roots):
+                level = 1 if depth == 2 and group < 4 else 0
+                factor = (n//128) * 2**level
+                expected += [(level, factor*x+dx, factor*y+dy,
+                              [x+(dx+.5)/factor, y+(dy+.5)/factor])
+                             for dy in range(factor) for dx in range(factor)]
+            self.assertEqual([(v['level'], v['i'], v['j'], v['center']) for v in p], expected)
+            for name in axis.GROUPS:
+                group = [v for v in p if v['group'] == name]
+                self.assertEqual(sum(v['area'] for v in group), 1)
+                np.testing.assert_array_equal(np.mean([v['center'] for v in group], axis=0),
+                                              np.array(axis.GROUPS[name])+.5)
+                for v in group:
+                    self.assertTrue(m[v['level']][v['j'], v['i']])
+            w = rc.workload(m, axis.case_steps(n)[-1])
+            work += w['active_collision_updates']
+            if depth == 2:
+                self.assertEqual(w['active_cells_by_level'], [12288, 16384])
+                self.assertEqual(w['level_hops'], [8, 16])
+            self.assertEqual(axis.case_steps(n), [r*(n//128) for r in range(9)])
+        self.assertEqual(work, 1900544)
+
+    def test_postg_moments_stress_and_matching_fine_control(self):
+        for level in (0, 1):
+            x, _ = centers((33, 41), level)
+            k, h, tau = 4*np.pi/128, 2.**-level, .5+3*.0384*2**level
+            uy = U*np.sin(k*x+np.pi/4)
+            eq = sw.equilibrium(np.ones_like(x), np.zeros_like(x), uy)
+            g = axis.initial(x.shape, level, 128, .0384)
+            a = 3*W[:, None, None]*CX[:, None, None]*CY[:, None, None]*h*k*U*np.cos(k*x+np.pi/4)
+            np.testing.assert_allclose(g, eq-(tau-1)*a, rtol=0, atol=6e-17)
+            for v in (np.ones(9), CX, CY):
+                np.testing.assert_allclose(np.einsum('q,qyx->yx', v, g-eq), 0, rtol=0, atol=2e-16)
+            np.testing.assert_allclose(np.einsum('q,qyx->yx', CX*CY, g-eq),
+                                       (1-tau)*h*k*U*np.cos(k*x+np.pi/4)/3, atol=6e-17)
+            self.assertGreater(g.min(), 0)
+        np.testing.assert_array_equal(axis.initial((256, 256), 0, 256, .0768),
+                                      axis.initial((256, 256), 1, 128, .0384))
+
+    def test_signed_momentum_and_actual_center_time(self):
+        m = masks(128, 2)
+        probes = axis.select(m)
+        trace = axis.new_trace(probes, .0384, [rc.AXIS_SQUARE, 128, 2, 'projected_linear'])
+        for step in axis.STEPS:
+            arrays = []
+            for level, mask in enumerate(m):
+                x, _ = centers(mask.shape, level)
+                uy, _ = axis.exact(x, step, 128, .0384)
+                arrays.append(sw.equilibrium(np.full_like(x, 1.001), np.zeros_like(x), uy))
+            trace['steps'].append(step)
+            trace['values'].append(axis.sample(arrays, probes))
+        trace['bytes'] = axis.validate(trace)
+        self.assertEqual(trace['bytes'], len(json.dumps(trace, allow_nan=False).encode()))
+        for row in trace['group_moments']:
+            for name, v in row['groups'].items():
+                self.assertAlmostEqual(v['drho'], .001, places=14)
+                x = np.array([p['center'][0] for p in probes if p['group'] == name])
+                uy, gradient = axis.exact(x, row['root_step'], 128, .0384)
+                self.assertAlmostEqual(v['djy'], .001*uy.mean(), places=14)
+                self.assertAlmostEqual(v['mean_duy_dx'], gradient.mean(), places=14)
+            self.assertAlmostEqual(row['groups']['inner_left']['mean_duy_dx'],
+                                   -row['groups']['inner_bottom']['mean_duy_dx'], places=14)
+        with patch.object(axis, 'TRACE_MAX_BYTES', trace['bytes']-1), self.assertRaises(ValueError):
+            axis.validate(trace)
+        for mutate in (lambda t: t['steps'].pop(),
+                       lambda t: t['steps'].__setitem__(1, 0),
+                       lambda t: t['values'][1].__setitem__(0, float('nan')),
+                       lambda t: t['probes'].pop()):
+            bad = deepcopy(trace); mutate(bad)
+            with self.assertRaises(ValueError): axis.validate(bad)
+
+    def test_four_full_mock_schedules_and_old_rejection(self):
+        helper = legacy.ProbeTraceTests()
+        helper.setUp()
+        def arrays(m, n, t):
+            return [axis.initial(v.shape, level, n, U*n/100) for level, v in enumerate(m)]
+        for _, n, depth, scheme in axis.CASES:
+            with patch.object(helper, 'exact_arrays', side_effect=arrays):
+                r, counts, files = helper.run_mocked(n, depth, rc.AXIS_SQUARE, scheme,
+                                                     masks(n, depth), initializer=axis.initial)
+            self.assertEqual(counts[-1], 8*(n//128))
+            self.assertEqual([v['step'] for v in r['rows']], [0, 8*(n//128)])
+            self.assertEqual(r['probe_trace']['steps'], list(range(9)))
+            self.assertEqual(r['probe_trace']['probes'], axis.select(masks(n, depth), n))
+            self.assertEqual(r['mode'], [2, 0])
+            self.assertEqual(r['phase'], np.pi/4)
+            self.assertEqual(r['initialization'], axis.INITIALIZATION)
+            self.assertEqual(r['limits'], {})
+            self.assertIsNone(r['screening_pass'])
+            self.assertFalse(r['qualification_eligible'])
+            self.assertEqual(files, ['probe.json'])
+        old = legacy.NestedPresetTests(); old.setUp()
+        records = old.nested_set(); records[-1]['diagnostic'] = rc.AXIS_SQUARE
+        self.assertEqual(old.mod.assess_nested(records, 'r')['status'], 'INVALID_SET')
+        records = [legacy.receipt(128, 1), legacy.receipt(256, 1), legacy.receipt(128, 2),
+                   legacy.receipt(128, 2, 'projected_linear')]
+        records[-1].update(diagnostic=rc.AXIS_SQUARE, initialization=axis.INITIALIZATION)
+        self.assertEqual(old.mod.assess(records, 'r')['status'], 'INVALID_SET')
+        self.assertNotIn(rc.AXIS_SQUARE, old.mod.PRESETS)
+        with patch.object(axis, 'sample', return_value=[float('nan')]*24):
+            r, _, _ = helper.run_mocked(128, 1, rc.AXIS_SQUARE, 'constant', masks(128, 1),
+                                        initializer=axis.initial)
+        self.assertIn('error', r['probe_trace'])
+        self.assertTrue(r['completed'])
+        self.assertEqual(len(r['rows']), 2)
+
+    def test_admission_and_module_pin(self):
+        from tests.validation.multigrid import check_projected_pairs as gate
+        self.assertEqual(hashlib.sha256(Path(axis.__file__).read_bytes()).hexdigest(), rc.SHEAR_AXIS_SQUARE_SHA256)
+        with tempfile.TemporaryDirectory() as tmp:
+            base = ['run_case.py', '--repo', str(REPO), '--expected-revision', 'f'*40,
+                    '--output', str(Path(tmp)/'case.json'), '--dry-run', '--re', '100',
+                    '--flow', 'shear_wave', '--diagnostic', rc.AXIS_SQUARE, '--probe-trace']
+            for n, depth, scheme, valid in [(n, d, s, True) for _, n, d, s in axis.CASES] + [
+                    (512, 1, 'constant', False), (256, 2, 'projected_linear', False), (128, 3, 'constant', False)]:
+                for extra in ([], ['--final-snapshot'], ['--early-pressure'], ['--flow', 'taylor_green'], ['--re', '1000']):
+                    argv = base + ['--n', str(n), '--depth', str(depth), '--explosion', scheme] + extra
+                    with patch.object(sys, 'argv', argv), redirect_stdout(io.StringIO()), patch('sys.stderr', io.StringIO()), \
+                         patch.object(rc, 'verify_source', return_value={}) as verify, patch.object(rc, 'runtime_source', return_value={}):
+                        if valid and not extra:
+                            self.assertEqual(rc.main(), 0)
+                            verify.assert_called_once_with(REPO, 'f'*40, 'shear_axis_square', probes=True)
+                        else:
+                            with self.assertRaises(SystemExit): rc.main()
+                            verify.assert_not_called()
+        with patch.object(gate, 'preflight', return_value=rc.NUMERICAL_REVISION), \
+             patch.object(rc, 'provenance', return_value=dict(source_tree_sha256=rc.SOURCE_SHA256)):
+            rc.verify_source(REPO, 'r', 'shear_axis_square', probes=True)
+            for module, key, value in ((axis, '__file__', '/wrong/helper.py'), (rc, 'SHEAR_AXIS_SQUARE_SHA256', 'wrong')):
+                with patch.object(module, key, value), self.assertRaisesRegex(RuntimeError, 'shear_axis_square'):
+                    rc.verify_source(REPO, 'r', 'shear_axis_square', probes=True)