From 1f263b9e4619bd8dd86673bbc8350713f24e8028 Mon Sep 17 00:00:00 2001 From: TRON Date: Sun, 18 Oct 2026 00:20:08 -0700 Subject: [PATCH] test: retain and localize channel GPU parity failures --- tests/unit/test_channel_parity_diagnostic.py | 47 ++++++++ .../channel_parity_diagnostic/README.md | 32 ++++++ .../channel_parity_diagnostic/discover.py | 104 ++++++++++++++++++ .../channel_parity_diagnostic/replay.py | 102 +++++++++++++++++ 4 files changed, 285 insertions(+) create mode 100644 tests/unit/test_channel_parity_diagnostic.py create mode 100644 tests/validation/multigrid/channel_parity_diagnostic/README.md create mode 100644 tests/validation/multigrid/channel_parity_diagnostic/discover.py create mode 100644 tests/validation/multigrid/channel_parity_diagnostic/replay.py diff --git a/tests/unit/test_channel_parity_diagnostic.py b/tests/unit/test_channel_parity_diagnostic.py new file mode 100644 index 0000000..3673eaa --- /dev/null +++ b/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 diff --git a/tests/validation/multigrid/channel_parity_diagnostic/README.md b/tests/validation/multigrid/channel_parity_diagnostic/README.md new file mode 100644 index 0000000..7427a32 --- /dev/null +++ b/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. diff --git a/tests/validation/multigrid/channel_parity_diagnostic/discover.py b/tests/validation/multigrid/channel_parity_diagnostic/discover.py new file mode 100644 index 0000000..95bd529 --- /dev/null +++ b/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()) diff --git a/tests/validation/multigrid/channel_parity_diagnostic/replay.py b/tests/validation/multigrid/channel_parity_diagnostic/replay.py new file mode 100644 index 0000000..695a078 --- /dev/null +++ b/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)