Spaces:
Paused
Paused
File size: 3,692 Bytes
07fb183 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """Fail-closed Wave-4 canonical cross-backend parity contract gate.
Hosted CI does not have physical CUDA. This gate verifies that CPU and CUDA
conform to the canonical ExecutionBackend surface, share one canonical
configuration identity, and that the named physical acceptance test is wired
through canonical parity functions rather than Playground helpers.
Physical D1/D2 execution is performed by scripts/run_cuda_hardware_acceptance.py.
"""
from __future__ import annotations
import ast
from collections.abc import Mapping
from pathlib import Path
from src.acceleration.cuda import CUDABackend
from src.acceleration.cuda.recurrent import (
CPUReferenceBackend,
recurrent_fixture,
recurrent_inputs_to_mapping,
)
from src.runtime.backend import ExecutionBackend
from src.verification.parity import config_fingerprint
ROOT = Path(__file__).resolve().parents[1]
TEST_FILE = ROOT / "tests" / "test_cuda_wave4_backend.py"
HARDWARE_TEST = "test_physical_cpu_cuda_cross_backend_d1_d2_parity"
def _hardware_test_source() -> str:
source = TEST_FILE.read_text(encoding="utf-8")
tree = ast.parse(source, filename=str(TEST_FILE))
for node in tree.body:
if (
isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name == HARDWARE_TEST
):
segment = ast.get_source_segment(source, node)
if segment is None:
raise RuntimeError("could not recover hardware acceptance test source")
return segment
raise RuntimeError(f"missing named hardware acceptance test: {HARDWARE_TEST}")
def failures() -> list[str]:
problems: list[str] = []
cpu: ExecutionBackend = CPUReferenceBackend()
cuda: ExecutionBackend = CUDABackend()
config = recurrent_inputs_to_mapping(
recurrent_fixture(n_neurons=8, ticks=4, model="lif")
)
seed = 12345
cpu.initialize(config, seed)
cuda.initialize(config, seed)
cpu_config = cpu.snapshot().payload.get("config")
cuda_config = cuda.snapshot().payload.get("config")
if not isinstance(cpu_config, Mapping) or not isinstance(cuda_config, Mapping):
problems.append("backend snapshots must expose canonical mapping configs")
elif config_fingerprint(cpu_config) != config_fingerprint(cuda_config):
problems.append("CPU/CUDA canonical configuration fingerprints differ")
try:
test_source = _hardware_test_source()
except RuntimeError as exc:
problems.append(str(exc))
test_source = ""
required_tokens = (
"CPUReferenceBackend",
"CUDABackend",
"exact_spike_parity",
"state_vector_parity",
"tolerance=1.0e-4",
)
for token in required_tokens:
if token not in test_source:
problems.append(
"hardware parity acceptance test is missing canonical token: " + token
)
full_test_source = TEST_FILE.read_text(encoding="utf-8")
if "MHRN_TEST_CUDA_HARDWARE" not in full_test_source:
problems.append("hardware parity acceptance test must remain opt-in")
if "playground" in test_source.lower():
problems.append(
"hardware parity acceptance test must not use Playground parity helpers"
)
return problems
def main() -> int:
problems = failures()
if problems:
raise SystemExit(
"Wave-4 canonical parity contract failed:\n" + "\n".join(problems)
)
print("Wave-4 canonical parity contract: PASS")
print("hardware acceptance test: PRESENT / OPT-IN")
print("D3 status: NOT_CLAIMED_IN_WAVE4")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|