Validate coarse preconditioner evidence and fallback comparisons

This commit is contained in:
2026-09-17 16:42:22 -04:00
parent 370dbbf73b
commit 48cb8c527f
2 changed files with 271 additions and 3 deletions

View File

@@ -127,6 +127,45 @@ class BenchmarkCheckerTests(unittest.TestCase):
change(data)
write_csv(path, data)
def refresh_output_hashes(self):
path = self.campaign / 'screen_run.json'
record = json.loads(path.read_text())
record['output_hashes'] = {p.name: checker.digest(p) for p in self.folder.iterdir()}
write_json(path, record)
def make_coarse(self, variant='coupled-shape-coarse', fallback=0, sweeps=1):
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))
active = 2 if fallback == 0 else 0
independent = 0 if fallback == 1 else 2
stable = {
'input_modes': 3, 'independent_modes': independent, 'active_modes': active,
'fallback': fallback, 'minimum_overlap_lower_bound': 'nan' if fallback == 1 else
1e-10 if fallback == 2 else 0.1,
'reciprocal_overlap_condition': 'nan' if fallback == 1 else 0.1,
'setup_jacobian_applications': 5, 'setup_inverse_applications': 2,
'setup_seconds': 0.15, 'persistent_vector_bytes': 128,
}
records = []
for stage, calls in [('constructed', 0), ('before_krylov', active + 1), ('after_krylov', active + 5)]:
metrics = stable | {'applications': calls, 'inverse_applications': calls, 'failed_applications': 0}
if stage == 'constructed':
metrics.update(physical_inverse_construction_seconds=0.7, basis_construction_seconds=0.1,
complete_coarse_construction_seconds=0.3)
records.extend({'role': 'candidate', 'variant': variant, 'stage': stage, 'metric': key, 'value': value}
for key, value in metrics.items())
records.append({'role': 'candidate', 'variant': variant, 'stage': 'validation',
'metric': 'maximum_exact_coarse_relative_error', 'value': 1e-12 if active else 0.0})
write_csv(self.folder / 'preconditioner_coarse.csv', records)
path = self.campaign / 'provenance.json'
provenance = json.loads(path.read_text())
provenance['cases'][0]['arguments'][-1] = variant
write_json(path, provenance)
self.refresh_output_hashes()
def assert_no_claim(self, result):
self.assertFalse(result['passed'])
self.assertFalse(result['comparison']['comparison_valid'])
@@ -259,6 +298,117 @@ class BenchmarkCheckerTests(unittest.TestCase):
self.assertFalse(result['passed'])
self.assert_no_claim(result['cases']['screen'])
def test_active_coarse_requires_real_subspace_and_complete_construction_cost(self):
self.make_coarse()
result = checker.inspect_campaign(self.campaign)
self.assertTrue(result['passed'], result)
comparison = result['cases']['screen']['comparison']
self.assertAlmostEqual(comparison['equal_gate_speedup'], 10.1 / 5.1)
coarse = comparison['coarse']['candidate']
self.assertEqual(coarse['active_modes'], 2)
self.assertEqual(coarse['exact_subspace_validation'], 'verified')
self.assertEqual(coarse['method_status'], 'active_coarse')
def test_invalid_coarse_identity_cost_or_work_cannot_support_speedup(self):
for metric, value, stage in [
('maximum_exact_coarse_relative_error', 2e-8, 'validation'),
('maximum_exact_coarse_relative_error', 'nan', 'validation'),
('physical_inverse_construction_seconds', 0.1, 'constructed'),
('basis_construction_seconds', 0.9, 'constructed'),
('applications', 2, 'before_krylov'),
('failed_applications', 1, 'after_krylov')]:
with self.subTest(metric=metric, value=value):
self.make_case()
self.make_coarse()
def corrupt(data):
for row in data:
if row['metric'] == metric and row['stage'] == stage:
row['value'] = value
self.alter_rows('preconditioner_coarse.csv', corrupt)
self.assert_no_claim(checker.inspect_case(self.folder))
def test_coarse_missing_duplicate_or_changing_preparation_is_rejected(self):
for corruption in ('missing', 'duplicate', 'changed'):
with self.subTest(corruption=corruption):
self.make_case()
self.make_coarse()
def corrupt(data):
if corruption == 'missing':
data.pop(0)
elif corruption == 'duplicate':
data.append(dict(data[0]))
else:
next(row for row in data if row['stage'] == 'after_krylov' and
row['metric'] == 'active_modes')['value'] = 0
self.alter_rows('preconditioner_coarse.csv', corrupt)
self.assert_no_claim(checker.inspect_case(self.folder))
def test_coarse_fallback_is_valid_but_empty_identity_proves_nothing(self):
for fallback in (1, 2):
with self.subTest(fallback=fallback):
self.make_case()
self.make_coarse(fallback=fallback)
result = checker.inspect_campaign(self.campaign)
self.assertTrue(result['passed'], result)
comparison = result['cases']['screen']['comparison']
self.assertIsNone(comparison['equal_gate_speedup'])
self.assertIsNone(comparison['effective_base_equal_gate_speedup'])
coarse = comparison['coarse']['candidate']
self.assertEqual(coarse['method_status'], 'baseline_fallback')
self.assertEqual(coarse['effective_method'], 'independent')
self.assertEqual(coarse['exact_subspace_validation'], 'not_applicable_empty_subspace')
json.dumps(result, allow_nan=False)
def test_schur_coarse_fallback_credits_only_the_actual_base(self):
for sweeps in (0, 1):
with self.subTest(sweeps=sweeps):
self.make_case()
self.make_coarse('gravity-schur-coarse', fallback=2, sweeps=sweeps)
result = checker.inspect_campaign(self.campaign)
self.assertTrue(result['passed'], result)
comparison = result['cases']['screen']['comparison']
self.assertIsNone(comparison['equal_gate_speedup'])
if sweeps:
self.assertAlmostEqual(comparison['effective_base_equal_gate_speedup'], 10.1 / 5.1)
self.assertEqual(comparison['coarse']['candidate']['effective_method'], 'gravity-schur')
else:
self.assertIsNone(comparison['effective_base_equal_gate_speedup'])
self.assertEqual(comparison['coarse']['candidate']['effective_method'], 'independent')
def test_contradictory_coarse_modes_or_fallback_are_rejected(self):
for metric, value in [('fallback', 1), ('fallback', 8), ('active_modes', 0), ('independent_modes', 4),
('minimum_overlap_lower_bound', 1e-10)]:
with self.subTest(metric=metric, value=value):
self.make_case()
self.make_coarse()
def corrupt(data):
for row in data:
if row['metric'] == metric:
row['value'] = value
self.alter_rows('preconditioner_coarse.csv', corrupt)
self.assert_no_claim(checker.inspect_case(self.folder))
def test_schur_fallback_does_not_claim_speedup_when_baseline_misses_gate(self):
self.make_case(baseline_defect=0.2)
self.make_coarse('gravity-schur-coarse', fallback=2)
result = checker.inspect_campaign(self.campaign)
self.assertTrue(result['passed'], result)
comparison = result['cases']['screen']['comparison']
self.assertEqual(comparison['outcome'], 'candidate_only_pass')
self.assertIsNone(comparison['equal_gate_speedup'])
self.assertIsNone(comparison['effective_base_equal_gate_speedup'])
def test_missing_coarse_hash_invalidates_even_a_schur_fallback_speedup(self):
self.make_coarse('gravity-schur-coarse', fallback=2)
path = self.campaign / 'screen_run.json'
record = json.loads(path.read_text())
record['output_hashes'].pop('preconditioner_coarse.csv')
write_json(path, record)
result = checker.inspect_campaign(self.campaign)
self.assertFalse(result['passed'])
self.assert_no_claim(result['cases']['screen'])
self.assertIsNone(result['cases']['screen']['comparison']['effective_base_equal_gate_speedup'])
if __name__ == '__main__':
unittest.main()