diff --git a/tests/validation/taylor_green/assess_projected_pairs.py b/tests/validation/taylor_green/assess_projected_pairs.py index 8d02595..2f93787 100644 --- a/tests/validation/taylor_green/assess_projected_pairs.py +++ b/tests/validation/taylor_green/assess_projected_pairs.py @@ -1,6 +1,7 @@ """New-source four-case repeat of 11901a7; frozen oracle and gates are unchanged.""" import argparse +import hashlib import json import math from pathlib import Path @@ -8,16 +9,25 @@ from pathlib import Path from assess import BASELINE_FACTOR, FLOORS from oracle import LIMITS, TIMES, U, rectangles from run_case import (INITIALIZATION, sample_steps, NUMERICAL_REVISION, SOURCE_SHA256, - BUILD_CAPTURE_SHA256, REFERENCE_HARNESS) + BUILD_CAPTURE_SHA256, REFERENCE_HARNESS, SCALED_DIAGNOSTIC) -# Host-generated snapshots of a135252 at the frozen two local configurations. +# Host-generated snapshots of a135252 at the frozen local configurations. # These describe expected inputs; actual sidecars are captured before cleanup. BUILD_SOURCES = { 128: "803177416b7599ab89cc5afd35cf4a28a673b2f4bb8e8a6a1abc0b6d77610873", 256: "17ec5e60f87b840d3d13b8778bc8a3aecf485469c5a061af46c5cc19b73396cf", + 512: "4e85e38098858ec95c137d1b1e96846a58d9cf3cc242b349a3ad7c8b8e3ce0aa", } CASES = ((100, 128, 1, "constant"), (100, 256, 1, "constant"), (100, 128, 2, "constant"), (100, 128, 2, "projected_linear")) +REFERENCE_REVISION = "2cf9853952e52f59bea63cf5c1d3a1cd76e53eed" +# Audited 2cf9853 scalar receipts, archived at evidence commit 90849a55cad4a0da9b72747da7bfe6eb75962a44. +REFERENCE_RECEIPTS = { + "uniform128": "724efafc267eaa59dcc59e8d4c16177b67f6c6a08fd2732a481833237bb19cee", + "uniform256": "3c312a99ebf747a135674dd481b77455a3d2c9a406ca388f30524b37dad9bbe9", + "constant128": "2ee28efd1e81b2e28b7902b6c4f95bb2c1ea4e5f0ec484638fa5ae83dca34bba", + "projected128": "ee67b1d4d90a048c0d3d1a16447c6eab3d89c94d151f791306e373f96dd8b963", +} NUMERICS = dict(lattice_model="D2Q9", collision="SRT", store_precision="FP32", ddf_shifting=True, streaming="double_buffer", les_enabled=False, boundary="periodic", block_shape=[8, 8], @@ -73,7 +83,11 @@ def valid_builds(record, n, depth): return True -def assess(records, expected_revision=None): +def assess(records, expected_revision=None, *, diagnostic=None): + if diagnostic not in (None, SCALED_DIAGNOSTIC): + raise ValueError("Unknown diagnostic") + base_n = 256 if diagnostic else 128 + cases = tuple((re, n * base_n // 128, depth, scheme) for re, n, depth, scheme in CASES) indexed, duplicates = {}, [] for record in records: key = tuple(record.get(k) for k in ("re_L", "n", "depth", "explosion")) @@ -81,9 +95,11 @@ def assess(records, expected_revision=None): duplicates.append(str(key)) else: indexed[key] = record - missing = [str(k) for k in CASES if k not in indexed] - checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(CASES), + missing = [str(k) for k in cases if k not in indexed] + checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(cases), no_duplicate_cases=not duplicates, finite_receipts=finite_tree(records)) + if diagnostic or any("diagnostic" in r for r in records): + checks["diagnostic_identity"] = all(r.get("diagnostic") == diagnostic for r in records) sources = [r.get("source", {}) for r in records] checks["same_source_and_harness"] = bool(sources) and len({ json.dumps(s, sort_keys=True) for s in sources}) == 1 @@ -106,9 +122,11 @@ def assess(records, expected_revision=None): ) and len(set(devices)) == 1 enabled = [r.get("early_pressure", {}).get("enabled") for r in records] checks["same_early_sampling"] = bool(enabled) and all(type(v) is bool for v in enabled) and len(set(enabled)) == 1 + if diagnostic: + checks["diagnostic_early_sampling"] = all(v is True for v in enabled) shared_valid = all(checks.values()) complete, case_valid, early = {}, {}, [] - for key in CASES: + for key in cases: if key not in indexed: continue r = indexed[key] @@ -135,7 +153,7 @@ def assess(records, expected_revision=None): (("module", "__init__.py"), ("compiler", "cuda/compiler_v2.py"), ("kernel_root", "lbm/kernels"))) expected_patch = [] if depth == 1 else rectangles(n, depth) checks[label + " patch"] = json.dumps(r.get("refinements")) == json.dumps(expected_patch) - cells = [n*n] if depth == 1 else [8704, 30720] + cells = [n*n] if depth == 1 else [17*n*n//32, 15*n*n//8] hops = [16*n * 2**l for l in range(depth)] work = r.get("workload", {}) checks[label + " workload"] = ( @@ -168,19 +186,22 @@ def assess(records, expected_revision=None): comparisons, scheme_results = [], {} for scheme in ("constant", "projected_linear"): - key = (100, 128, 2, scheme) + key = (100, base_n, 2, scheme) for metric, floor in FLOORS.items(): - amr, coarse, fine = (worst(k, metric) for k in (key, CASES[0], CASES[1])) + amr, coarse, fine = (worst(k, metric) for k in (key, cases[0], cases[1])) budget = None if coarse is None else max(BASELINE_FACTOR * coarse, floor) budget = budget if finite(budget) else None ratio = None if amr is None or fine is None else amr / max(fine, 1e-30) passed = amr is not None and budget is not None and amr <= budget checks[f"baseline {scheme}/{metric}"] = passed comparisons.append(dict(explosion=scheme, metric=metric, amr=amr, - uniform128=coarse, baseline_budget=budget, passed=passed, - uniform256=fine, amr_to_uniform256=ratio if finite(ratio) else None)) + baseline_budget=budget, passed=passed, + **{f"uniform{base_n}": coarse, f"uniform{2*base_n}": fine, + f"amr_to_uniform{2*base_n}": ratio if finite(ratio) else None})) + if diagnostic: + comparisons[-1]["floor_dominates"] = None if coarse is None else floor >= BASELINE_FACTOR * coarse baseline_passed = all(checks[f"baseline {scheme}/{metric}"] for metric in FLOORS) - valid = shared_valid and all(case_valid.get(k, False) for k in (CASES[0], CASES[1], key)) + valid = shared_valid and all(case_valid.get(k, False) for k in (cases[0], cases[1], key)) scheme_results[scheme] = dict(passed=valid and baseline_passed, valid_case_and_controls=valid, baseline_passed=baseline_passed) return dict(passed=all(checks.values()), missing=missing, duplicates=duplicates, checks=checks, @@ -190,19 +211,89 @@ def assess(records, expected_revision=None): floors=FLOORS, baseline_factor=BASELINE_FACTOR, main_statistic="maximum of four post-initial samples at t*=.12,.24,.36,.48", early_statistic="sampled pressure maximum only, t*=k*.001875, k=0..64; not an all-step or continuous maximum", - finest_control="uniform256 and AMR/ uniform256 ratios are descriptive, not acceptance gates", + finest_control=f"uniform{2*base_n} and AMR/ uniform{2*base_n} ratios are descriptive, not acceptance gates", scope="2D periodic Re100 smooth-decay diagnostic; no new algorithm, convergence-order or general AMR qualification") +def load_reference(paths): + """Use the four audited historical files verbatim, never relabel them fresh controls.""" + blobs = [path.read_bytes() for path in paths] + hashes = [hashlib.sha256(blob).hexdigest() for blob in blobs] + if len(hashes) != 4 or set(hashes) != set(REFERENCE_RECEIPTS.values()): + raise ValueError("Historical N128 receipt hashes differ from the audited 2cf9853 evidence") + return [json.loads(blob) for blob in blobs] + + +def resolution_trend(records, reference, expected_revision): + """Signed changes are descriptive; they never replace the frozen acceptance budgets.""" + old = assess(reference, REFERENCE_REVISION) + new = assess(records, expected_revision, diagnostic=SCALED_DIAGNOSTIC) + # Numerical screen failures still have useful raw errors; only malformed inputs block trends. + valid = all(v for result in (old, new) for k, v in result["checks"].items() + if not k.startswith("baseline ") and not k.endswith(" absolute")) + valid = valid and all(maximum(r["rows"], m) is not None for r in reference + records for m in LIMITS) + oracle = reference[0]["source"]["harness_sha256"].get("oracle.py") if valid else None + valid = valid and bool(oracle) and records[0]["source"]["harness_sha256"].get("oracle.py") == oracle + if not valid: + return dict(valid=False, per_time=[], worst_post_initial=[], uniform256_bridge=[]) + key = lambda r: (r["re_L"], r["n"], r["depth"], r["explosion"]) + before, after = {key(r): r for r in reference}, {key(r): r for r in records} + + def change(role, metric, low, high, **extra): + ratio = high / low if low else None + percent = 100*(ratio-1) if ratio is not None else None + return dict(role=role, metric=metric, n128_campaign_error=low, n256_campaign_error=high, + signed_change=high-low, new_to_old=ratio if finite(ratio) else None, + ratio_unavailable_reason=None if finite(ratio) else "zero historical error" if not low else "overflow", + change_percent=percent if finite(percent) else None, **extra) + + per_time, worst, bridge = [], [], [] + for old_key, role in zip(CASES, ("uniform_root", "uniform_fine", "constant", "projected_linear")): + re, n, depth, scheme = old_key + low, high = before[old_key], after[(re, 2*n, depth, scheme)] + for metric in LIMITS: + per_time.extend(change(role, metric, a[metric], b[metric], tstar=a["tstar"]) + for a, b in zip(low["rows"], high["rows"])) + worst.append(change(role, metric, maximum(low["rows"][1:], metric), + maximum(high["rows"][1:], metric))) + for a, b in zip(before[CASES[1]]["rows"], after[CASES[1]]["rows"]): + bridge.extend(dict(metric=m, tstar=a["tstar"], historical=a[m], fresh=b[m], signed_change=b[m]-a[m]) + for m in LIMITS) + low, high = before[CASES[1]], after[CASES[1]] + bridge_identity = dict(main_rows_identical=low["rows"] == high["rows"], + early_pressure_rows_identical=low["early_pressure"]["rows"] == high["early_pressure"]["rows"]) + same_runtime = reference[0]["runtime_identity"] == records[0]["runtime_identity"] + return dict(valid=True, reference_revision=REFERENCE_REVISION, reference_receipt_sha256=REFERENCE_RECEIPTS, + reference_scheme_results=old["scheme_results"], per_time=per_time, worst_post_initial=worst, + uniform256_bridge=bridge, uniform256_bridge_identity=bridge_identity, + same_recorded_runtime=same_runtime, + trend_interpretation_ready=all(bridge_identity.values()) and same_runtime, + scope="Historical same-production/oracle reference versus fresh controls; same-placement two-resolution trend only, not convergence order or phase robustness") + + if __name__ == "__main__": p = argparse.ArgumentParser(description=__doc__) p.add_argument("files", type=Path, nargs="+") p.add_argument("--output", type=Path, required=True) p.add_argument("--expected-revision", required=True) + p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC]) + p.add_argument("--reference-files", type=Path, nargs=4) args = p.parse_args() if args.output.exists(): p.error("Output already exists; choose a new path to retain prior evidence") - result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision) + if bool(args.diagnostic) != bool(args.reference_files): + p.error("--diagnostic scaled_n256 requires --reference-files; references are diagnostic-only") + records = [json.loads(path.read_text()) for path in args.files] + result = assess(records, args.expected_revision, diagnostic=args.diagnostic) + if args.diagnostic: + try: + reference = load_reference(args.reference_files) + except ValueError as exc: + p.error(str(exc)) + result["diagnostic"] = args.diagnostic + result["resolution_trend"] = resolution_trend(records, reference, args.expected_revision) + result["checks"]["valid_historical_reference_and_trend"] = result["resolution_trend"]["valid"] + result["passed"] = all(result["checks"].values()) args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n") print(json.dumps(result, indent=2, allow_nan=False)) raise SystemExit(0 if result["passed"] else 1) diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index 2f4954c..edba615 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -18,6 +18,7 @@ SOURCE_SHA256 = "a5dbfd804befce868067d79a23e5fd47dce7f505b3f9b1bb2145fb57cb5f217 BUILD_CAPTURE_SHA256 = "3dbaa6547cb0a4cb1541054f3b64ad8b6e18cc70527ac6166e56b944e9bd464a" REFERENCE_HARNESS = "11901a7f77557a901e017a171eeae1758d584fa5" INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p" +SCALED_DIAGNOSTIC = "scaled_n256" def sample_steps(n, early_pressure=False): @@ -145,10 +146,12 @@ def main(): p.add_argument("--run-gpu", action="store_true") p.add_argument("--dry-run", action="store_true") p.add_argument("--re", type=int, choices=[100, 1000], required=True) - p.add_argument("--n", type=int, choices=[32, 64, 128, 256], required=True) + p.add_argument("--n", type=int, choices=[32, 64, 128, 256, 512], required=True) p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True) p.add_argument("--output", type=Path, required=True) p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant") + p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC], + help="explicit four-case N256/fine512 same-placement diagnostic") p.add_argument("--early-pressure", action="store_true", help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64") args = p.parse_args() @@ -156,6 +159,13 @@ def main(): p.error("Output already exists; choose a new path to retain prior evidence") if args.explosion != "constant" and args.depth != 2: p.error("projected_linear requires depth 2") + if args.diagnostic: + cases = {(256, 1, "constant"), (512, 1, "constant"), + (256, 2, "constant"), (256, 2, "projected_linear")} + if args.re != 100 or (args.n, args.depth, args.explosion) not in cases or not args.early_pressure: + p.error("scaled_n256 requires one of its four Re100 cases and --early-pressure") + elif args.n == 512: + p.error("N512 requires --diagnostic scaled_n256") selected = os.environ.get("CUDA_VISIBLE_DEVICES", "") if not args.dry_run and (not args.run_gpu or not selected or "," in selected or selected == "-1"): p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required") @@ -164,7 +174,8 @@ def main(): source = verify_source(repo, args.expected_revision) runtime = runtime_source(repo) if args.dry_run: - print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True))) + print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True, + **({"diagnostic": args.diagnostic} if args.diagnostic else {})))) return 0 from CelerisLab.cuda.context import CudaContext from CelerisLab.lbm.blocks import BlockLayout2D @@ -212,6 +223,8 @@ def main(): sampled_max_pressure_l2=None), completed=False, ) + if args.diagnostic: + report["diagnostic"] = args.diagnostic args.output.parent.mkdir(parents=True, exist_ok=True) def save(): temporary = args.output.with_suffix(args.output.suffix + ".tmp") diff --git a/tests/validation/taylor_green/test_projected_pairs.py b/tests/validation/taylor_green/test_projected_pairs.py index 50b9f8b..d723ccd 100644 --- a/tests/validation/taylor_green/test_projected_pairs.py +++ b/tests/validation/taylor_green/test_projected_pairs.py @@ -17,6 +17,7 @@ from unittest.mock import patch import numpy as np import run_case +import assess_projected_pairs as assessor from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION, SOURCE_SHA256, assess) from oracle import LIMITS, TIMES, U, rectangles @@ -31,7 +32,7 @@ RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0 def source(): return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256, - harness_sha256={"run_case.py": "a" * 64}, + harness_sha256={"run_case.py": "a" * 64, "oracle.py": "c" * 64}, build_capture_sha256=run_case.BUILD_CAPTURE_SHA256) @@ -42,12 +43,14 @@ def build_metadata(size): arch="sm_70", flags=["--use_fast_math"], ptx_sha256="b" * 64) -def records(early=False): +def records(early=False, diagnostic=None): from CelerisLab.lbm.hierarchy import build_hierarchy from CelerisLab.lbm.refinement import ACTIVE out = [] for re, n, depth, scheme in CASES: + if diagnostic: + n *= 2 cfg = run_case.config(n, re) refinements = [] if depth == 1 else rectangles(n, depth) h = build_hierarchy(cfg, refinements) @@ -66,6 +69,8 @@ def records(early=False): early_pressure=dict(enabled=early, rows=[ dict(step=s, tstar=s*U/n, pressure_l2=.01 if s not in main_steps else 0.0) for s in early_steps], sampled_max_pressure_l2=.01 if early else None))) + if diagnostic: + out[-1]["diagnostic"] = diagnostic return out @@ -251,7 +256,7 @@ class ProjectedPairsTests(unittest.TestCase): def test_pinned_build_sources_match_host_generated_inputs(self): from CelerisLab.cuda import compiler_v2 - for size in (128, 256): + for size in (128, 256, 512): cfg = replace(run_case.config(size, 100), nx=10, ny=10) cfg.validate() root = compiler_v2.generate_config(cfg) @@ -263,6 +268,111 @@ class ProjectedPairsTests(unittest.TestCase): finally: compiler_v2.cleanup_build_dir(root) + def test_scaled_geometry_work_and_frozen_gates(self): + data = records(True, run_case.SCALED_DIAGNOSTIC) + self.assertEqual(sum(r["workload"]["active_collision_updates"] for r in data), 4714397696) + self.assertEqual([r["viscosity"] for r in data], [.0768, .1536, .0768, .0768]) + self.assertEqual(data[-1]["tau_by_level"], [.7303999999999999, .9608]) + for r in data: + self.assertEqual([row["tstar"] for row in r["rows"]], TIMES) + self.assertEqual(len(r["early_pressure"]["rows"]), 65) + if r["depth"] == 2: + self.assertEqual(r["refinements"], [[(32, 32, 224, 192)]]) + self.assertEqual(r["workload"]["active_cells_by_level"], [34816, 122880]) + self.assertEqual(r["workload"]["level_hops"], [4096, 8192]) + result = assess(data, MOCK_REVISION, diagnostic=run_case.SCALED_DIAGNOSTIC) + self.assertTrue(result["passed"]) + self.assertEqual(result["floors"], dict(velocity_l2=.002, pressure_l2=.02, energy_relative=.004)) + self.assertTrue(all(row["floor_dominates"] for row in result["comparisons"])) + data[-1]["rows"][-1]["velocity_l2"] = .00200001 + self.assertFalse(assess(data, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"]) + self.assertFalse(assess(data)["passed"]) + data[-1]["rows"][-1]["velocity_l2"] = 0.0 + for mutate in (lambda r: r.pop("diagnostic"), lambda r: r.update(diagnostic="wrong"), + lambda r: r["source"].update(revision="wrong"), + lambda r: r["early_pressure"].update(enabled=False), + lambda r: r.update(refinements=[[(32, 32, 224, 200)]])): + bad = deepcopy(data) + mutate(bad[-1]) + self.assertFalse(assess(bad, MOCK_REVISION, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"]) + + def test_raw_resolution_changes_and_reference_identity(self): + old, new = records(True), records(True, run_case.SCALED_DIAGNOSTIC) + for r in old: + r["source"]["revision"] = assessor.REFERENCE_REVISION + old[-2]["rows"][-1]["velocity_l2"] = .004 # historical constant failure remains usable + old[-1]["rows"][-1]["velocity_l2"] = .001 + new[-1]["rows"][-1]["velocity_l2"] = .0015 # floor pass, but worse raw error + result = assessor.resolution_trend(new, old, MOCK_REVISION) + self.assertTrue(result["valid"]) + self.assertTrue(result["trend_interpretation_ready"]) + self.assertFalse(result["reference_scheme_results"]["constant"]["passed"]) + self.assertEqual((len(result["per_time"]), len(result["worst_post_initial"]), + len(result["uniform256_bridge"])), (80, 16, 20)) + row = next(r for r in result["per_time"] if r["role"] == "projected_linear" + and r["metric"] == "velocity_l2" and r["tstar"] == .48) + self.assertEqual((row["signed_change"], row["new_to_old"], row["change_percent"]), (.0005, 1.5, 50.)) + self.assertTrue(assess(new, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"]) + self.assertIsNone(result["per_time"][0]["new_to_old"]) # zero denominator is explicit + self.assertEqual(result["per_time"][0]["ratio_unavailable_reason"], "zero historical error") + new[-1]["rows"][-1]["velocity_l2"] = .06 # retain raw errors even when absolute screen fails + self.assertTrue(assessor.resolution_trend(new, old, MOCK_REVISION)["valid"]) + self.assertFalse(assess(new, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"]) + new[-1]["rows"][-1]["velocity_l2"] = .0015 + for sample in ("rows", "early_pressure"): + bad = deepcopy(new) + if sample == "rows": + bad[0]["rows"][-1]["velocity_l2"] = .0001 + else: + bad[0]["early_pressure"]["rows"][1]["pressure_l2"] = .009 + result = assessor.resolution_trend(bad, old, MOCK_REVISION) + self.assertTrue(result["valid"]) + self.assertFalse(result["trend_interpretation_ready"]) + for mutate in (lambda r: r["source"].update(revision=MOCK_REVISION), + lambda r: r["source"].update(source_tree_sha256="wrong"), + lambda r: r["source"]["harness_sha256"].update(**{"oracle.py": "wrong"}), + lambda r: r.update(diagnostic=run_case.SCALED_DIAGNOSTIC), + lambda r: r.update(refinements=[]), lambda r: r["rows"].pop()): + bad = deepcopy(old) + mutate(bad[-1]) + self.assertFalse(assessor.resolution_trend(new, bad, MOCK_REVISION)["valid"]) + with tempfile.TemporaryDirectory() as tmp: + paths = [Path(tmp) / f"{i}.json" for i in range(4)] + for path, receipt in zip(paths, old): + path.write_text(json.dumps(receipt)) + with self.assertRaisesRegex(ValueError, "receipt hashes"): + assessor.load_reference(paths) + witnesses = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in paths} + with patch.object(assessor, "REFERENCE_RECEIPTS", witnesses): + self.assertEqual(assessor.load_reference(paths), json.loads(json.dumps(old))) + with self.assertRaises(ValueError): + assessor.load_reference(paths[:-1] + paths[:1]) + paths[-1].write_text(paths[-1].read_text() + " ") + with self.assertRaises(ValueError): + assessor.load_reference(paths) + + def test_scaled_runner_requires_explicit_bounded_diagnostic(self): + with tempfile.TemporaryDirectory() as tmp: + base = ["run_case.py", "--repo", str(REPO), "--expected-revision", MOCK_REVISION, + "--output", str(Path(tmp) / "case.json"), "--dry-run"] + diagnostic = ["--diagnostic", run_case.SCALED_DIAGNOSTIC, "--early-pressure"] + for n, depth, re, extra, valid in ( + (512, 1, 100, [], False), (512, 1, 100, diagnostic, True), + (256, 2, 100, diagnostic, True), (128, 2, 100, diagnostic, False), + (256, 3, 100, diagnostic, False), (256, 2, 1000, diagnostic, False), + (256, 2, 100, diagnostic[:2], False), (512, 2, 100, diagnostic, False)): + argv = base + ["--n", str(n), "--depth", str(depth), "--re", str(re)] + extra + with self.subTest(argv=argv), patch.object(sys, "argv", argv), \ + patch.object(run_case, "verify_source", return_value=source()) as verify, \ + redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()): + if valid: + self.assertEqual(run_case.main(), 0) + else: + with self.assertRaises(SystemExit) as caught: + run_case.main() + self.assertEqual(caught.exception.code, 2) + verify.assert_not_called() + def test_source_preflight_requires_clean_exact_revision_and_tree(self): import oracle from tests.validation.multigrid import check_projected_pairs as gate @@ -349,9 +459,15 @@ class ProjectedPairsTests(unittest.TestCase): def test_mocked_runner_default_early_and_failure(self): # Replace the GPU context and steppers before import. No CUDA module or flow is used. - for early, depth, fail in ((False, 1, None), (False, 2, None), (True, 2, None), + cases = [(early, depth, fail, 128, None, "projected_linear" if early else "constant") + for early, depth, fail in + ((False, 1, None), (False, 2, None), (True, 2, None), (True, 2, "resource"), (True, 2, "nonfinite"), - (True, 2, "probe_nonfinite")): + (True, 2, "probe_nonfinite"))] + cases += [(True, depth, None, n, run_case.SCALED_DIAGNOSTIC, scheme) + for n, depth, scheme in ((256, 1, "constant"), (512, 1, "constant"), + (256, 2, "constant"), (256, 2, "projected_linear"))] + for early, depth, fail, n, diagnostic, scheme in cases: calls, selectors = [], [] class FakeRun: @@ -395,11 +511,14 @@ class ProjectedPairsTests(unittest.TestCase): with tempfile.TemporaryDirectory() as tmp: output = Path(tmp) / "case.json" - argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128", + argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", str(n), "--depth", str(depth), "--output", str(output), "--expected-revision", MOCK_REVISION, "--run-gpu"] + argv += ["--explosion", scheme] if early: - argv += ["--early-pressure", "--explosion", "projected_linear"] + argv += ["--early-pressure"] + if diagnostic: + argv += ["--diagnostic", diagnostic] with patch.dict(sys.modules, modules), patch.object(sys, "argv", argv), \ patch.dict("os.environ", CUDA_VISIBLE_DEVICES="mock"), \ patch.object(run_case, "verify_source", return_value=source()) as verify, \ @@ -413,15 +532,16 @@ class ProjectedPairsTests(unittest.TestCase): self.assertEqual(run_case.main(), 0) receipt = json.loads(output.read_text()) verify.assert_called_once_with(REPO, MOCK_REVISION) - self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else []) + self.assertEqual(selectors, [scheme] if depth == 2 else []) self.assertEqual(receipt["source"], source()) self.assertEqual(receipt["runtime_identity"], RUNTIME) self.assertEqual(receipt["completed"], not fail) + self.assertEqual(receipt.get("diagnostic"), diagnostic) self.assertEqual(calls[:2], [0, 1]) # READY probe is part of this trajectory if not fail: self.assertEqual([r["tstar"] for r in receipt["rows"]], TIMES) self.assertEqual(len(receipt["early_pressure"]["rows"]), 65 if early else 0) - self.assertEqual(calls[-1], 2048) + self.assertEqual(calls[-1], 16*n) else: self.assertEqual(len(receipt["rows"]), 1) self.assertEqual(len(receipt["early_pressure"]["rows"]),