#!/usr/bin/env python3 """Small synthetic linear experiments testing whether a speedup is supportable. No executable, FEM setup, or solver is launched. The fixture uses J=I and keeps the exact solution/defect relationship while varying convergence and evidence. """ import csv import json from pathlib import Path import tempfile import unittest import check_preconditioner_benchmark as checker def write_csv(path, records): with path.open('w', newline='') as stream: writer = csv.DictWriter(stream, fieldnames=list(records[0])) writer.writeheader() writer.writerows(records) def write_vector(path, values): path.write_text(str(len(values)) + '\n' + '\n'.join(repr(v) for v in values) + '\n') def write_json(path, value): path.write_text(json.dumps(value, indent=2) + '\n') class BenchmarkCheckerTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.campaign = Path(self.temporary.name).resolve() self.folder = self.campaign / 'screen' self.folder.mkdir() self.rhs = [3.0, 4.0] self.make_case() def make_case(self, baseline_defect=0.1, candidate_defect=0.05): layout = [{'kind': 'residual', 'block': 'equilibrium', 'offset': 0, 'size': 2, 'scale': 1}, {'kind': 'value', 'block': 'state', 'offset': 0, 'size': 2, 'scale': 1}] write_csv(self.folder / 'layout.csv', layout) summary, blocks, actions, correction_blocks, history = [], [], [], [], [] for role, variant, defect, cost in ( ('baseline', 'independent', baseline_defect, 10.0), ('candidate', 'stellar-ldu', candidate_defect, 5.0)): # The saved defect convention is Jp-rhs, hence p=rhs+defect for J=I. solution = [self.rhs[0] + defect, self.rhs[1]] exact_defect = [defect, 0.0] for coordinates in ('normalized', 'physical'): write_vector(self.folder / f'{role}_{coordinates}_correction.txt', solution) write_vector(self.folder / f'{role}_{coordinates}_linear_defect.txt', exact_defect) summary.append({ 'role': role, 'variant': variant, 'restart': 10, 'maximum_iterations': 20, 'relative_tolerance': 0.03, 'rhs_norm': 5.0, 'absolute_target': 0.15, 'true_residual_norm': defect, 'true_relative': defect / 5.0, 'true_gate_pass': int(defect <= 0.15), 'solver_converged': int(defect <= 0.15), 'reported_initial_norm': 5.0, 'normalized_correction_norm': checker.norm(solution), 'iterations': 4, 'preconditioner_applications': 4, 'construction_seconds': 1.0, 'normalization_binding_seconds': 0.1, 'krylov_setup_seconds': 0.2, 'krylov_seconds': cost - 1.3, 'setup_plus_solve_seconds': cost, 'true_check_seconds': 0.1, 'setup_solve_and_true_check_seconds': cost + 0.1, 'jacobian_seconds': 1.0, 'preconditioner_seconds': 1.0, 'true_check_jacobian_applications': 1, }) blocks.append({'role': role, 'variant': variant, 'block': 'equilibrium', 'normalized_rhs_norm': 5.0, 'normalized_defect_norm': defect}) actions.append({'role': role, 'variant': variant, 'context_equivalence_required': int(role == 'baseline'), 'context_action_relative_difference': 0.0}) correction_blocks.append({'role': role, 'variant': variant, 'block': 'state', 'normalized_norm': checker.norm(solution), 'physical_norm': checker.norm(solution)}) history.append({'role': role, 'variant': variant, 'iteration': 4, 'reported_norm': defect, 'final': 1}) for file, data in [('preconditioner_summary.csv', summary), ('preconditioner_blocks.csv', blocks), ('preconditioner_actions.csv', actions), ('preconditioner_history.csv', history), ('preconditioner_correction_blocks.csv', correction_blocks)]: write_csv(self.folder / file, data) invariants = [] for role in ('context_reference', 'baseline', 'candidate'): for metric in ('accepted_physical_bitwise_identical', 'accepted_normalized_bitwise_identical', 'accepted_residual_bitwise_identical', 'correction_bitwise_identical', 'trial_state_bitwise_identical', 'trial_residual_bitwise_identical', 'candidate_physical_bitwise_identical', 'state_factors_bitwise_identical', 'residual_factors_bitwise_identical', 'preparation_generation_unchanged', 'normalized_operator_current', 'jacobian_action_bitwise_identical'): invariants.append({'role': role, 'metric': metric, 'value': 1}) invariants.append({'role': role, 'metric': 'jacobian_action_difference_norm', 'value': 0}) write_csv(self.folder / 'preconditioner_invariants.csv', invariants) write_csv(self.folder / 'preconditioner_reference.csv', [{'metric': 'context_inverse_correction_norm', 'value': 5.0}]) for name, values in { 'benchmark_normalized_rhs.txt': self.rhs, 'benchmark_normalized_base.txt': [2.0, 3.0], 'benchmark_physical_base.txt': [2.0, 3.0], 'benchmark_state_factors.txt': [1.0, 1.0], 'benchmark_residual_factors.txt': [1.0, 1.0], 'benchmark_invariant_direction.txt': [1.0, 0.0], 'benchmark_invariant_action.txt': [1.0, 0.0], 'context_reference_normalized_correction.txt': self.rhs, }.items(): write_vector(self.folder / name, values) (self.folder / 'metadata.txt').write_text('n=1\nJ=0\ncentral_phase=0\npreconditioner_probe=1\n') (self.folder / 'preconditioner_scope.txt').write_text('Synthetic J=I accuracy-gate comparison.\n') self.snapshot = self.campaign / 'snapshot.txt' write_vector(self.snapshot, [2.0, 3.0]) self.mesh = self.campaign / 'mesh.smesh' self.mesh.write_text('Synthetic fixture; never opened by a solver.\n') executable = self.campaign / 'plateau_diagnostic' executable.write_bytes(b'Synthetic executable provenance; not executable.\n') write_json(self.campaign / 'provenance.json', { 'commit': 'synthetic-test', 'source': str(self.campaign), 'binary_sha256': checker.digest(executable), 'mesh': str(self.mesh), 'mesh_sha256': checker.digest(self.mesh), 'cases': [{ 'id': 'screen', 'arguments': ['--probe-state', str(self.snapshot), '--preconditioner-variant', 'stellar-ldu']}], }) write_json(self.campaign / 'screen_run.json', { 'exit_code': 0, 'finished_utc': '2026-09-18T00:00:00Z', 'input_state_hashes': {str(self.snapshot): checker.digest(self.snapshot)}, 'output_hashes': {p.name: checker.digest(p) for p in self.folder.iterdir()}, }) def alter_rows(self, filename, change): path = self.folder / filename data = checker.rows(path) change(data) write_csv(path, data) def assert_no_claim(self, result): self.assertFalse(result['passed']) self.assertFalse(result['comparison']['comparison_valid']) self.assertIsNone(result['comparison']['equal_gate_speedup']) def test_matched_verified_gates_allow_measured_speedup(self): result = checker.inspect_campaign(self.campaign) self.assertTrue(result['passed'], result) comparison = result['cases']['screen']['comparison'] self.assertEqual(comparison['outcome'], 'both_pass') self.assertAlmostEqual(comparison['equal_gate_speedup'], 10.1 / 5.1) def test_failed_gates_are_valid_observations_without_speedup(self): for baseline, candidate, outcome in ((0.2, 0.3, 'neither_passes'), (0.2, 0.1, 'candidate_only_pass'), (0.1, 0.2, 'baseline_only_pass')): with self.subTest(outcome=outcome): self.make_case(baseline, candidate) result = checker.inspect_campaign(self.campaign) self.assertTrue(result['passed'], result) comparison = result['cases']['screen']['comparison'] self.assertEqual(comparison['outcome'], outcome) self.assertIsNone(comparison['equal_gate_speedup']) def test_exact_zero_defect_does_not_invent_a_defect_ratio(self): self.make_case(0.0, 0.0) result = checker.inspect_case(self.folder) self.assertTrue(result['passed'], result) self.assertIsNone(result['comparison']['candidate_over_baseline_defect']) self.assertIsNotNone(result['comparison']['equal_gate_speedup']) json.dumps(result, allow_nan=False) def test_missing_invariant_cannot_support_speedup(self): self.alter_rows('preconditioner_invariants.csv', lambda data: data.pop(0)) self.assert_no_claim(checker.inspect_case(self.folder)) def test_duplicate_invariant_cannot_hide_a_failed_check(self): def corrupt(data): duplicate = dict(data[0]) data[0]['value'] = 0 data.append(duplicate) self.alter_rows('preconditioner_invariants.csv', corrupt) self.assert_no_claim(checker.inspect_case(self.folder)) def test_false_reported_gate_is_rejected(self): self.make_case(0.2, 0.1) self.alter_rows('preconditioner_summary.csv', lambda data: data[0].update(true_gate_pass=1)) self.assert_no_claim(checker.inspect_case(self.folder)) def test_faster_total_cannot_omit_verification_or_setup(self): self.alter_rows('preconditioner_summary.csv', lambda data: data[1].update(setup_solve_and_true_check_seconds=0.01)) self.assert_no_claim(checker.inspect_case(self.folder)) def test_saved_defect_and_csv_must_agree(self): write_vector(self.folder / 'candidate_normalized_linear_defect.txt', [1.0, 0.0]) self.assert_no_claim(checker.inspect_case(self.folder)) def test_same_vector_norm_does_not_hide_wrong_block_distribution(self): layout = [ {'kind': 'residual', 'block': 'gravity', 'offset': 0, 'size': 1, 'scale': 1}, {'kind': 'residual', 'block': 'material', 'offset': 1, 'size': 1, 'scale': 1}, ] write_csv(self.folder / 'layout.csv', layout) block_rows = [] for role, defect in [('baseline', 0.1), ('candidate', 0.05)]: block_rows.extend([{'role': role, 'variant': 'unused', 'block': 'gravity', 'normalized_rhs_norm': 3.0, 'normalized_defect_norm': defect}, {'role': role, 'variant': 'unused', 'block': 'material', 'normalized_rhs_norm': 4.0, 'normalized_defect_norm': 0.0}]) write_csv(self.folder / 'preconditioner_blocks.csv', block_rows) write_vector(self.folder / 'candidate_normalized_linear_defect.txt', [0.0, 0.05]) self.assert_no_claim(checker.inspect_case(self.folder)) def test_default_equivalence_cannot_be_disabled_in_csv(self): self.alter_rows('preconditioner_actions.csv', lambda data: data[0].update(context_equivalence_required=0)) self.assert_no_claim(checker.inspect_case(self.folder)) def test_zero_sweep_gravity_control_requires_default_equivalence(self): for variant, sweeps, equivalent in (('gravity-schur', 0, True), ('gravity-schur', 1, False), ('stellar-ldu-gravity-schur', 0, False)): with self.subTest(variant=variant, sweeps=sweeps): self.make_case() def summary(data): data[0]['schur_correction_sweeps'] = 0 data[1].update(variant=variant, schur_correction_sweeps=sweeps) self.alter_rows('preconditioner_summary.csv', summary) self.alter_rows('preconditioner_actions.csv', lambda data: data[1].update(variant=variant, context_equivalence_required=int(equivalent))) result = checker.inspect_case(self.folder) self.assertTrue(result['passed'], result) if equivalent: self.alter_rows('preconditioner_actions.csv', lambda data: data[1].update(context_equivalence_required=0)) self.assert_no_claim(checker.inspect_case(self.folder)) def test_omitted_input_or_output_hash_cannot_support_speedup(self): for key, remove in [('input_state_hashes', str(self.snapshot)), ('output_hashes', 'candidate_normalized_linear_defect.txt')]: with self.subTest(key=key): self.make_case() path = self.campaign / 'screen_run.json' record = json.loads(path.read_text()) record[key].pop(remove) write_json(path, record) result = checker.inspect_campaign(self.campaign) self.assertFalse(result['passed']) self.assert_no_claim(result['cases']['screen']) def test_changed_provenance_input_executable_mesh_or_output_blocks_claim(self): for name in ('input', 'executable', 'mesh', 'output'): with self.subTest(name=name): self.make_case() path = {'input': self.snapshot, 'executable': self.campaign / 'plateau_diagnostic', 'mesh': self.mesh, 'output': self.folder / 'preconditioner_summary.csv'}[name] with path.open('a') as stream: stream.write('\n') result = checker.inspect_campaign(self.campaign) self.assertFalse(result['passed']) self.assert_no_claim(result['cases']['screen']) def test_in_progress_record_is_not_a_completed_comparison(self): path = self.campaign / 'screen_run.json' record = json.loads(path.read_text()) record.pop('finished_utc') write_json(path, record) result = checker.inspect_campaign(self.campaign) self.assertFalse(result['passed']) self.assert_no_claim(result['cases']['screen']) if __name__ == '__main__': unittest.main()