test: retain and localize channel GPU parity failures

tests/unit/test_channel_parity_diagnostic.py

@@ -0,0 +1,47 @@
+"""CPU-only traceback capture checks; no CUDA or GPU test import."""
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+import numpy as np
+
+PATH = Path(__file__).resolve().parents[1] / 'validation/multigrid/channel_parity_diagnostic/discover.py'
+spec = importlib.util.spec_from_file_location('channel_discovery', PATH)
+discovery = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(discovery)
+
+
+def test_captures_exact_original_frame_and_arrays():
+    def test_uniform_matched_root():
+        cfg = SimpleNamespace(velocity=.025)
+        shifted, profile, clamp, damping, block = True, 'parabolic', False, .8, (4, 1)
+        expected = np.array([.001], np.float32)
+        actual = np.nextafter(expected, np.float32(1))
+        np.testing.assert_array_equal(actual, expected)
+    try:
+        test_uniform_matched_root()
+    except AssertionError as exc:
+        config, expected, actual = discovery.capture_case(exc)
+    assert config['shifted'] == 1 and config['profile'] == 'parabolic'
+    assert config['clamp'] == 0 and config['damping'] == .8
+    assert config['block'] == [4, 1] and config['steps'] == 8
+    np.testing.assert_array_equal(actual, np.nextafter(expected, np.float32(1)))
+    assert expected.flags.owndata and actual.flags.owndata
+
+
+def test_other_assertion_does_not_guess_case():
+    def test_uniform_matched_root():
+        cfg = SimpleNamespace(velocity=.025)
+        shifted, profile, clamp, damping, block = False, 'uniform', True, .5, (20, 24)
+        expected = actual = np.array([np.nan], np.float32)
+        assert np.isfinite(expected).all()
+    try:
+        test_uniform_matched_root()
+    except AssertionError as exc:
+        assert discovery.capture_case(exc) is None
+
+
+def test_unrelated_array_assertion_does_not_guess_case():
+    try:
+        np.testing.assert_array_equal([1], [2])
+    except AssertionError as exc:
+        assert discovery.capture_case(exc) is None

tests/validation/multigrid/channel_parity_diagnostic/README.md

@@ -0,0 +1,32 @@
+# Root-channel parity diagnostic
+
+This is a read-only diagnostic, not a replacement acceptance gate. It requires a
+caller-admitted GPU and the unchanged checkout containing
+`test_channel_root_gpu.py`. It does not run unrelated body/Jet tests.
+
+From the checkout, run:
+
+```
+python /path/to/channel_parity_diagnostic/discover.py --checkout "$PWD" --output /new/evidence/directory
+```
+
+Discovery invokes only the original `test_uniform_matched_root` method, with its
+original loops and exact assertion. It stops at the first assertion failure,
+extracts the configuration and arrays from that original traceback, saves the
+failure verbatim, and serially starts `replay.py` in a fresh Python process.
+The original failure remains a failure: exit 1 means the assertion was captured,
+even if the targeted replay succeeds or does not reproduce it. Exit 2 means the
+original test did not reproduce the failure; exit 3 means a different assertion
+failed and no configuration was guessed. Other errors are retained and raised.
+
+Targeted replay compiles only the uniform and one failed blocked configuration,
+serially, caching the two modules. It saves PTX and compiler metadata, all
+numeric and bit-only mismatches with coordinates/hex/ULPs at each step, first
+numeric divergence arrays, same-input localization, and comparison against the
+original multistep call. Signed-zero-only bit changes cannot trigger the first
+numeric divergence. The final replay consistency checks establish whether the
+one-step observation sequence agrees with the original call structure.
+
+Outputs retain the original source revision and gate SHA256. No source,
+assertion, tolerance, compiler flags, collision math, or kernel scheduling is
+changed. Findings must be reviewed before any production fix or gate change.

tests/validation/multigrid/channel_parity_diagnostic/discover.py

@@ -0,0 +1,104 @@
+"""Discover the first failure of the unchanged matched-root GPU test, then replay.
+
+This script never runs the body/Jet test, changes the solver, or changes its
+assertions. GPU admission remains the caller's responsibility.
+"""
+import argparse
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import sys
+import traceback
+
+
+def capture_case(exc):
+    """Extract the original test frame; no inferred profile/block/configuration."""
+    frames = []
+    tb = exc.__traceback__
+    while tb is not None:
+        frames.append(tb.tb_frame)
+        tb = tb.tb_next
+    caller = next((f for f in frames if f.f_code.co_name == 'test_uniform_matched_root'), None)
+    if caller is None or not any(f.f_code.co_name == 'assert_array_equal' for f in frames):
+        return None
+    local = caller.f_locals
+    required = ('cfg', 'shifted', 'profile', 'clamp', 'damping', 'block', 'expected', 'actual')
+    if not all(key in local for key in required):
+        return None
+    cfg = local['cfg']
+    return dict(shifted=int(local['shifted']), profile=local['profile'],
+                clamp=int(local['clamp']), damping=float(local['damping']),
+                block=[int(v) for v in local['block']], steps=8,
+                config_repr=repr(cfg)), local['expected'].copy(), local['actual'].copy()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--checkout', type=Path, default=Path.cwd())
+    parser.add_argument('--output', type=Path, required=True, help='New directory; existing evidence is never overwritten')
+    args = parser.parse_args()
+    checkout, output = args.checkout.resolve(), args.output.resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    import os
+    os.chdir(checkout)
+    sys.path.insert(0, str(checkout / 'src'))
+    import numpy as np
+    gate_path = checkout / 'tests/validation/multigrid/test_channel_root_gpu.py'
+    spec = importlib.util.spec_from_file_location('original_channel_gpu_gate', gate_path)
+    gate = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(gate)
+    revision = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True, text=True, check=True).stdout.strip()
+    manifest = dict(checkout=str(checkout), revision=revision,
+                    gate_sha256=hashlib.sha256(gate_path.read_bytes()).hexdigest(),
+                    test='ChannelRootGPU.test_uniform_matched_root')
+    (output / 'manifest.json').write_text(json.dumps(manifest, indent=2))
+    test = gate.ChannelRootGPU('test_uniform_matched_root')
+    try:
+        # The original method and np.testing.assert_array_equal are unchanged.
+        # Its existing loops naturally stop at the first failing assertion.
+        test.test_uniform_matched_root()
+    except AssertionError as exc:
+        original_trace = traceback.format_exc()
+        (output / 'original_failure.txt').write_text(original_trace)
+        print(original_trace, file=sys.stderr, flush=True)
+        case = capture_case(exc)
+        if case is None:
+            (output / 'status.json').write_text(json.dumps(dict(status='different_assertion_failure', replay_started=False), indent=2))
+            return 3
+        config, expected, actual = case
+        (output / 'failed_case.json').write_text(json.dumps(config, indent=2))
+        np.savez(output / 'original_failure_arrays.npz', expected=expected, actual=actual)
+        print('CAPTURED_CASE ' + json.dumps(config), flush=True)
+    except Exception:
+        (output / 'discovery_error.txt').write_text(traceback.format_exc())
+        (output / 'status.json').write_text(json.dumps(dict(status='discovery_error', replay_started=False), indent=2))
+        raise
+    else:
+        (output / 'status.json').write_text(json.dumps(dict(status='original_failure_not_reproduced', replay_started=False), indent=2))
+        print('The original matched-root test completed without reproducing the failure. No targeted replay was started.', flush=True)
+        return 2
+
+    # Discovery's context has closed. Run targeted diagnostics in a fresh process,
+    # serially, never concurrently with another compilation or GPU run here.
+    replay = Path(__file__).resolve().with_name('replay.py')
+    command = [sys.executable, str(replay), '--shifted', str(config['shifted']),
+               '--profile', config['profile'], '--clamp', str(config['clamp']),
+               '--damping', str(config['damping']), '--block', *map(str, config['block']),
+               '--steps', str(config['steps']), '--output', str(output / 'targeted')]
+    (output / 'replay_command.json').write_text(json.dumps(command, indent=2))
+    with (output / 'targeted_stdout.txt').open('w') as stdout, (output / 'targeted_stderr.txt').open('w') as stderr:
+        result = subprocess.run(command, cwd=checkout, stdout=stdout, stderr=stderr)
+    targeted = output / 'targeted/result.json'
+    details = json.loads(targeted.read_text()) if targeted.exists() else None
+    status = dict(status='original_assertion_failure_retained', replay_started=True,
+                  replay_returncode=result.returncode, replay_result=details)
+    (output / 'status.json').write_text(json.dumps(status, indent=2))
+    print(json.dumps(status), flush=True)
+    # Preserve the original failed gate even if replay fails or does not reproduce.
+    return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

tests/validation/multigrid/channel_parity_diagnostic/replay.py

@@ -0,0 +1,102 @@
+"""Read-only production-kernel parity diagnostic; no altered math or thresholds.
+Run in checkout with PYTHONPATH=src. Supply the exact failed subcase.
+"""
+import argparse
+import importlib.util
+import json
+from pathlib import Path
+import shutil
+import sys
+import numpy as np
+
+p = argparse.ArgumentParser()
+p.add_argument('--shifted', type=int, choices=(0, 1), required=True)
+p.add_argument('--profile', choices=('uniform', 'parabolic'), required=True)
+p.add_argument('--clamp', type=int, choices=(0, 1), required=True)
+p.add_argument('--damping', type=float, required=True)
+p.add_argument('--block', type=int, nargs=2, required=True, metavar=('BY', 'BX'))
+p.add_argument('--steps', type=int, default=8)
+p.add_argument('--output', type=Path, required=True)
+a = p.parse_args()
+a.output.mkdir(parents=True, exist_ok=False)
+sys.path.insert(0, str(Path('src').resolve()))
+path = Path('tests/validation/multigrid/test_channel_root_gpu.py').resolve()
+spec = importlib.util.spec_from_file_location('channel_gpu_gate', path)
+gate = importlib.util.module_from_spec(spec); spec.loader.exec_module(gate)
+
+# Preserve the exact NVCC-generated PTX before the existing compiler cleans up.
+original_compile = gate.compiler.compile_kernel
+original_module = gate.compile_module
+cache = {}
+def retain_compile(*args, **kwargs):
+    result = original_compile(*args, **kwargs)
+    source = kwargs.get('source', 'kernel_v2.cu')
+    shutil.copyfile(result, a.output / (Path(source).stem + '.ptx'))
+    metadata = Path(result + '.json')
+    if metadata.exists(): shutil.copyfile(metadata, a.output / (Path(source).stem + '.ptx.json'))
+    return result
+def cached_module(cfg, context, source):
+    cfg.validate()
+    key = (repr(cfg), source)
+    if key not in cache: cache[key] = original_module(cfg, context, source)
+    return cache[key]
+gate.compiler.compile_kernel = retain_compile
+gate.compile_module = cached_module
+
+cfg = gate.LBMConfig(nx=24, ny=20, viscosity=.1, velocity=.025,
+                    ddf_shifting=bool(a.shifted), inlet_profile=a.profile,
+                    outlet_backflow_clamp=bool(a.clamp), outlet_srt_neq_damp=a.damping)
+config = dict(shifted=a.shifted, profile=a.profile, clamp=a.clamp,
+              damping=a.damping, block=a.block, steps=a.steps)
+(a.output / 'config.json').write_text(json.dumps(config, indent=2))
+def ordered(bits):
+    return int((~bits & 0xffffffff) if bits & 0x80000000 else (bits | 0x80000000))
+def report(expected, actual, step, label):
+    ub = expected.view(np.uint32); ab = actual.view(np.uint32)
+    indices = np.argwhere(ub != ab)
+    numeric = ~((expected == actual) | (np.isnan(expected) & np.isnan(actual)))
+    numeric_count = int(np.count_nonzero(numeric))
+    rows = []
+    for q, y, x in indices:
+        u, b = int(ub[q,y,x]), int(ab[q,y,x])
+        category = ('wall_ghost' if y in (0,cfg.ny-1) else 'inlet_ghost' if x==0
+                    else 'outlet_ghost' if x==cfg.nx-1 else 'wall_adjacent_fluid' if y in (1,cfg.ny-2) else 'bulk_fluid')
+        rows.append(dict(numeric_mismatch=bool(numeric[q,y,x]), q=int(q), y=int(y), x=int(x), category=category,
+                         uniform=float(expected[q,y,x]), blocked=float(actual[q,y,x]),
+                         uniform_hex=f'{u:08x}', blocked_hex=f'{b:08x}',
+                         ulps=abs(ordered(u)-ordered(b))))
+    record = dict(label=label, step=step, bit_count=len(rows), numeric_count=numeric_count,
+                  max_abs=float(np.max(np.abs(actual.astype(float)-expected.astype(float)))),
+                  max_ulps=max((r['ulps'] for r in rows), default=0),
+                  all_finite=bool(np.isfinite(expected).all() and np.isfinite(actual).all()),
+                  differences=rows)
+    (a.output / f'{label}_{step:03d}.json').write_text(json.dumps(record,indent=2))
+    print(json.dumps({k:v for k,v in record.items() if k!='differences'}), flush=True)
+    if rows: print(json.dumps(rows[:12]), flush=True)
+    return numeric_count > 0
+
+with gate.CudaContext(0) as context:
+    initial = gate._encode_physical(gate.state(cfg.ny,cfg.nx), bool(a.shifted))
+    uniform, blocked = initial.copy(), initial.copy()
+    found = False
+    for step in range(1,a.steps+1):
+        previous_uniform, previous_blocked = uniform, blocked
+        uniform = gate.run_root(cfg, context, previous_uniform, steps=1)
+        blocked = gate.run_root(cfg, context, previous_blocked, tuple(a.block), steps=1)
+        differs = report(uniform, blocked, step, 'propagated')
+        if differs and not found:
+            found = True
+            same_input_blocked = gate.run_root(cfg, context, previous_uniform, tuple(a.block), steps=1)
+            report(uniform, same_input_blocked, step, 'same_input')
+            np.savez(a.output/'first_divergence.npz', initial=initial,
+                     previous_uniform=previous_uniform, previous_blocked=previous_blocked,
+                     uniform=uniform, blocked=blocked, same_input_blocked=same_input_blocked)
+    # Verify repeated one-step diagnostics still reproduce the original eight-hop call.
+    direct_uniform = gate.run_root(cfg, context, initial, steps=a.steps)
+    direct_blocked = gate.run_root(cfg, context, initial, tuple(a.block), steps=a.steps)
+    reproduced = report(direct_uniform, direct_blocked, a.steps, 'original_multistep')
+    report(direct_uniform, uniform, a.steps, 'uniform_replay_consistency')
+    report(direct_blocked, blocked, a.steps, 'blocked_replay_consistency')
+
+(a.output / 'result.json').write_text(json.dumps(dict(numeric_failure_reproduced=reproduced, first_numeric_divergence_found=found), indent=2))
+print(json.dumps(dict(numeric_failure_reproduced=reproduced)), flush=True)