#!/usr/bin/env python3 """Small synthetic searches exercise certificates, selection and provenance.""" import csv import json import math from pathlib import Path import tempfile import unittest import check_damped_coupled_trial as c from test_coupled_trial_checker import csv_file, vec_file class DampedCheckerTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name).resolve(); self.folder = self.root / 'synthetic'; self.folder.mkdir() self.make_case() def make_case(self, kind='damped'): for path in self.folder.iterdir(): path.unlink() pfx, folder, nan = c.P, self.folder, math.nan mode = 'near' if kind in ('near', 'raw_failure') else ('none' if kind == 'none' else ('full' if kind in ('full', 'budget', 'linear_failure', 'cap') else 'damped')) x, p, f = [1., 2., 3.], [-1., -1., -1.], [1., 1., 1.] sf, rf = [2.] * 3, [3.] * 3 save = lambda name, value: vec_file(folder / (pfx + name + '.txt'), value) for name, values in {'base': x, 'base_physical': [v / 2 for v in x], 'base_residual': f, 'direction': p, 'base_action': p, 'state_factors': sf, 'residual_factors': rf, 'restored_residual': f, 'restored_action': p}.items(): save(name, values) layout = [{'kind': typ, 'block': name, 'offset': i, 'size': 1} for typ, names in ( ('value', ('density', 'shape', 'enthalpy')), ('residual', ('closure', 'shape', 'hydrostatic'))) for i, name in enumerate(names)] csv_file(folder / (pfx + 'layout.csv'), layout) max_steps = 1 if mode in ('none', 'full') else 4 cap_ratio = .01 if kind == 'cap' else 1. records, candidates, evaluations = [], [], [] attempts = iterations = deferred = 0 plans = ['attempt'] if kind == 'none': plans = ['accepted'] elif kind in ('full', 'linear_failure', 'cap', 'best'): plans += ['accepted'] elif kind == 'budget': plans += ['budget', 'accepted'] elif kind == 'near': plans = ['deferred', 'attempt'] elif kind == 'raw_failure': plans = ['raw_failure', 'accepted'] for i, plan in enumerate(plans): alpha, suffix = 2. ** -i, '_a' + str(i) raw = [v + alpha * w for v, w in zip(x, p)]; prediction = [1. - alpha] * 3 save('raw' + suffix, raw); save('prediction' + suffix, prediction) usable = plan != 'raw_failure'; attempted = plan == 'attempt'; before_accepted = plan == 'accepted' before_f = [.5] * 3 if before_accepted else ([3., 0., 0.] if plan == 'deferred' else [2., 0., 0.]) if usable: save('raw_residual' + suffix, before_f); save('before' + suffix, raw); save('before_residual' + suffix, before_f) status = {'accepted': 'not_attempted' if mode == 'none' else 'original_accepted', 'budget': 'budget_exhausted', 'deferred': 'prediction_deferred', 'raw_failure': 'not_attempted'}.get(plan, 'kept') selected, selected_f, selected_beta = raw, before_f, nan core, count, preps, restores = -1, 0, 0, 0 correction = [-.5, 0., 0.]; rhs = [a - b for a, b in zip(prediction, before_f)] action = [0., 0., 0.] if kind == 'linear_failure' else rhs defect = c.minus(action, rhs); gate = c.norm(defect) <= .03 * c.norm(rhs) if attempted: attempts += 1; iterations += 2 core = 1 if kind == 'linear_failure' else (2 if kind == 'cap' else 0) for name, values in {'rhs': rhs, 'correction': correction, 'action': action, 'defect': defect}.items(): save(name + suffix, values) if core: status = 'linear_failure' if core == 1 else 'norm_limit' else: proposal_norms = [2.1] if mode == 'full' else ([1.8, 1.9, 2., 2.1] if kind == 'best' else [2.1, 1.]) observed = [] for k, value in enumerate(proposal_norms): beta = .5 ** k; state = [v + beta * w for v, w in zip(raw, correction)]; residual = [value, 0., 0.] failed = kind == 'candidate_failure' and k == 0 cm = .5 * value * value; accepted = cm <= (1 - 1.e-4 * alpha) * 1.5 candidates.append({'trial': i, 'candidate': k, 'beta': beta, 'status': 1 if failed else 0, 'merit': nan if failed else cm, 'accepted': int(accepted and not failed), 'preparation_seconds': .05}) count += 1; preps += 1 if not failed: observed.append((beta, state, residual)) if kind == 'best': selected_beta = 1.; selected = [v + w for v, w in zip(raw, correction)]; selected_f = [1.8, 0., 0.] observed.append((1., selected, selected_f)); restores = 1; preps += 1 elif mode == 'full': status = 'not_improved'; observed.append((0., raw, before_f)); restores = 1; preps += 1 else: selected_beta, selected, selected_f = observed[-1] for e, (beta, state, residual) in enumerate(observed): tag = suffix + '_e' + str(e) save('evaluated_state' + tag, state); save('evaluated_residual' + tag, residual) cm = c.merit(residual) evaluations.append({'trial': i, 'evaluation': e, 'beta': beta, 'residual_norm': c.norm(residual), 'merit': cm, 'accepted': int(cm <= (1 - 1.e-4 * alpha) * 1.5)}) deferred += plan == 'deferred' accepted_now = usable and c.merit(selected_f) <= (1 - 1.e-4 * alpha) * 1.5 if usable: save('selected' + suffix, selected); save('selected_residual' + suffix, selected_f) if accepted_now: vec_file(folder / ('coupled_trial_selected_physical' + suffix + '.txt'), [v / 2 for v in selected]) gate_used = mode == 'near' and usable and not before_accepted and plan != 'budget' _, decrease, excess = c.eligibility(1.5, c.merit(prediction), c.merit(before_f), 1.) solve, verification = (.1, .01) if attempted else (0., 0.) refresh, prep_seconds = (.02 * (1 + restores) if attempted else 0.), .05 * preps correction_seconds = solve + verification + refresh + prep_seconds + (.01 if attempted else 0.) records.append({'trial': i, 'alpha': alpha, 'raw_prepared': int(usable), 'raw_reason': -1 if usable else 2, 'raw_stage': -1 if usable else 6, 'gravity_gate': 1, 'usable': int(usable), 'before_norm': c.norm(before_f) if usable else nan, 'before_merit': c.merit(before_f) if usable else nan, 'before_accepted': int(before_accepted), 'status': c.STATUS.index(status), 'attempted': int(attempted), 'attempts_total': attempts, 'aux_iterations': 2 if attempted else 0, 'aux_gate': int(attempted and gate), 'aux_threshold': .03 * c.norm(rhs) if attempted else nan, 'aux_defect': c.norm(defect) if attempted else nan, 'correction_norm': .5 if attempted else nan, 'correction_cap': cap_ratio * alpha * math.sqrt(3) if attempted else nan, 'selected_beta': selected_beta, 'candidate_count': count, 'predicted_decrease': decrease if gate_used else nan, 'excess_ratio': excess if gate_used else nan, 'selected_prepared': int(usable), 'selected_norm': c.norm(selected_f) if usable else nan, 'selected_merit': c.merit(selected_f) if usable else nan, 'selected_accepted': int(accepted_now), 'selected_original': int(usable and selected == raw), 'raw_seconds': .1, 'gravity_seconds': 0., 'correction_seconds': correction_seconds, 'preparations': preps, 'restorations': restores, 'solve_seconds': solve, 'refresh_seconds': refresh, 'verification_seconds': verification, 'preparation_seconds': prep_seconds, 'aux_core_status': core, 'aux_status': 0 if attempted else -1, 'aux_reported_gate': int(attempted), 'aux_rhs_norm': c.norm(rhs) if attempted else nan, 'aux_reported_true_norm': 0. if attempted else nan, 'aux_verification_applied': int(attempted), 'rhs_saved': int(attempted), 'correction_saved': int(attempted), 'verified_vectors_saved': int(attempted), 'gravity_target_defect': nan, 'gravity_target_threshold': nan, 'gravity_affine_disagreement': nan}) csv_file(folder / (pfx + 'trials.csv'), records) csv_file(folder / (pfx + 'candidates.csv'), candidates, ['trial', 'candidate', 'beta', 'status', 'merit', 'accepted', 'preparation_seconds']) csv_file(folder / (pfx + 'evaluations.csv'), evaluations, ['trial', 'evaluation', 'beta', 'residual_norm', 'merit', 'accepted']) last = records[-1] summary = {'schema_version': 1, 'mode': mode, 'gravity_path': 0, 'attempt_budget': 1, 'maximum_trials': 13, 'maximum_iterations': 160, 'restart': 160, 'maximum_beta_steps': max_steps, 'beta_reduction': .5, 'maximum_excess_ratio': 1., 'sufficient_decrease': 1.e-4, 'relative_tolerance': .03, 'absolute_tolerance': 1.e-14, 'maximum_correction_ratio': cap_ratio, 'accepted_commits': 0, 'shared_iteration_budget': 1, 'base_norm': math.sqrt(3), 'base_merit': 1.5, 'outer_relative_defect': 0., 'outer_linear_tolerance': .03, 'outer_verified_gate': 1, 'capture_seconds': .1, 'search_wall_seconds': 10., 'search_operation_seconds': sum(row['raw_seconds'] + row['correction_seconds'] for row in records), 'accepted_trial': len(records) - 1, 'accepted_alpha': last['alpha'], 'accepted_norm': last['selected_norm'], 'accepted_beta': last['selected_beta'], 'attempts': attempts, 'aux_iterations': iterations, 'deferred_attempts': deferred, 'probe_exception': 0, 'accepted_buffers_bitwise_identical': 1, 'direction_buffers_bitwise_identical': 1, 'normalization_factors_bitwise_identical': 1, 'restored_residual_error': 0., 'restored_action_error': 0., 'fresh_operator_and_inverse_current': 1} summary.update({'status_' + name: c.STATUS.index(name) for name in ('original_accepted', 'budget_exhausted', 'kept', 'prediction_deferred')}) summary.update({'candidate_status_' + name: i for i, name in enumerate(('evaluated', 'inadmissible', 'non_finite'))}) csv_file(folder / (pfx + 'summary.csv'), [{'metric': key, 'value': value} for key, value in summary.items()]) metadata = {'n': 1.5, 'K': 1., 'M': 1., 'J': 0., 'central_phase': 0, 'damped_coupled_probe': 1, 'coupled_trial_alphas_are_independent': 0, 'coupled_trial_admissibility': 'sampled_enthalpy_without_repair', 'coupled_search': mode, 'coupled_beta_steps': 4, 'coupled_beta_reduction': .5, 'coupled_excess_ratio': 1., 'linear_tol': .03, 'coupled_tolerance': .03, 'coupled_iterations': 160, 'coupled_restart': 160, 'coupled_maximum_ratio': cap_ratio, 'coupled_attempts_per_iteration': 1, 'strategy': 'coupled'} (folder / 'metadata.txt').write_text(''.join(f'{key}={value}\n' for key, value in metadata.items())) self.physical(last) self.make_campaign(mode) def physical(self, selected): records = [] def add(sample, source, q, key, value): records.append({'sample': sample, 'source': source, 'quadrature': q, 'metric': key, 'value': value}) for key in ('accepted_buffers_unchanged_before_restoration', 'accepted_buffers_unchanged_after_restoration', 'normalization_factors_unchanged', 'direction_buffers_unchanged', 'accepted_operator_and_inverse_current'): add(-1, 'restoration', 0, key, 1) add(-1, 'replay', 0, 'selected_snapshots_measured', 1) i = selected['trial']; source = f'coupled_trial_selected_physical_a{i}.txt' add(i, source, 0, 'fresh_preparation_passed', 1) for q in (14, 18): values = {'quadrature_order': q, 'polytropic_index': 1.5, 'polytropic_constant': 1., 'target_mass': 1., 'stellar_samples': 10, 'valid_stellar_samples': 10, 'invalid_stellar_samples': 0, 'negative_density_samples': 0, 'negative_enthalpy_samples': 0, 'invalid_density_eos_samples': 0, 'invalid_enthalpy_eos_samples': 0, 'volume': 1., 'complete_volume_measurement': 1, 'pressure_from_enthalpy_defined_everywhere': 1, 'pressure_from_density_defined_everywhere': 1, 'saved_prepared_l2_difference': 0., 'saved_prepared_maximum_difference': 0., 'saved_prepared_maximum_scaled_difference': 0., 'normalized_bordered_residual': selected['selected_norm'], 'normalized_unbordered_residual': selected['selected_norm'], 'normalized_border_action': 0., 'central_border': 0., 'mass': 1., 'mass_absolute_error': 0., 'minimum_density': 0., 'minimum_enthalpy': 0., 'binding_energy': -3., 'force_binding_energy': -3., 'kinetic_energy': 0., 'pressure_integral': 1., 'enthalpy_pressure_integral': 1.} values.update({prefix + force + 'virial_' + name: 0. for prefix in ('', 'enthalpy_') for force in ('', 'force_') for name in ('signed', 'error')}) for key, value in values.items(): add(i, source, q, key, value) csv_file(self.folder / 'coupled_trial_physical.csv', records) def make_campaign(self, mode): root = self.root state, direction, mesh = root / 'state.txt', root / 'direction.txt', root / 'mesh' vec_file(state, [.5, 1., 1.5]); vec_file(direction, [-1.] * 3); mesh.write_text('synthetic mesh') (root / 'plateau_diagnostic').write_text('synthetic executable') arguments = ['--damped-coupled-probe', '--coupled-search', mode, '--strategy', 'coupled', '--no-central', '--probe-state', str(state), '--probe-direction', str(direction)] cases = [{'id': 'synthetic', 'arguments': arguments}] (root / 'cases.json').write_text(json.dumps(cases)) (root / 'provenance.json').write_text(json.dumps({'cases': cases, 'commit': 'synthetic', 'source': str(root), 'mesh': str(mesh), 'mesh_sha256': c.digest(mesh), 'binary_sha256': c.digest(root / 'plateau_diagnostic')})) (root / 'synthetic_run.json').write_text(json.dumps({'id': 'synthetic', 'finished_utc': 'synthetic', 'exit_code': 0, 'command': [str(root / 'plateau_diagnostic'), *arguments, '--mesh', str(mesh), '--output', str(self.folder)], 'input_state_hashes': {str(path): c.digest(path) for path in (state, direction)}, 'output_hashes': {path.name: c.digest(path) for path in self.folder.iterdir()}})) def audit(self): return c.inspect_case(self.folder) def assert_passes(self): result = self.audit(); self.assertTrue(result['passed'], [row for row in result['checks'] if not row['passed']]) def mutate_row(self, file, key, value, row=0): path = self.folder / (c.P + file + '.csv'); records = c.rows(path); records[row][key] = value; csv_file(path, records) def add_gravity_path(self): folder = self.folder; delta = [0., .01, .02] for name in ('before_a0', 'selected_a0', 'evaluated_state_a0_e0', 'evaluated_state_a0_e1'): path = folder / (c.P + name + '.txt'); values = c.vector(path) vec_file(path, [a + b for a, b in zip(values, delta)]) vec_file(folder / (c.P + 'gravity_correction_a0.txt'), delta) vec_file(folder / (c.P + 'raw_residual_a0.txt'), [2., .2, -.1]) selected = c.vector(folder / (c.P + 'selected_a0.txt')) vec_file(folder / 'coupled_trial_selected_physical_a0.txt', [value / 2 for value in selected]) layout = [{'kind': kind, 'block': name, 'offset': i, 'size': 1} for kind, names in ( ('value', ('density', 'gravity_gradient', 'gravity_potential')), ('residual', ('closure', 'gravity_gradient_relation', 'poisson_balance'))) for i, name in enumerate(names)] csv_file(folder / (c.P + 'layout.csv'), layout) for key, value in (('gravity_target_defect', 0.), ('gravity_target_threshold', 1.e-12), ('gravity_affine_disagreement', 0.)): self.mutate_row('trials', key, value) summary = c.rows(folder / (c.P + 'summary.csv')) next(row for row in summary if row['metric'] == 'gravity_path')['value'] = 1 csv_file(folder / (c.P + 'summary.csv'), summary) path = folder / 'metadata.txt'; path.write_text(path.read_text().replace('strategy=coupled', 'strategy=curvature-coupled')) def test_damped_half_step(self): self.assert_passes() def test_near_gate_preserves_budget(self): self.make_case('near'); self.assert_passes() def test_full_fallback(self): self.make_case('full'); self.assert_passes() def test_shared_budget_exhaustion(self): self.make_case('budget'); self.assert_passes() def test_best_beta_replay(self): self.make_case('best'); self.assert_passes() def test_unprepared_beta_skipped(self): self.make_case('candidate_failure'); self.assert_passes() def test_no_correction_control(self): self.make_case('none'); self.assert_passes() def test_pressure_failure_preserves_budget(self): self.make_case('raw_failure'); self.assert_passes() def test_failed_full_true_gate(self): self.make_case('linear_failure'); self.assert_passes() def test_full_cap_rejects_before_damping(self): self.make_case('cap'); self.assert_passes() def test_gravity_path_uses_full_coupled_target(self): self.add_gravity_path(); self.assert_passes() def test_corrupt_gravity_target_fails(self): self.add_gravity_path(); self.mutate_row('trials', 'gravity_target_defect', 1.); self.assertFalse(self.audit()['passed']) def test_provenance_passes(self): self.assertTrue(c.inspect_campaign(self.root)['passed']) def test_missing_true_action_fails(self): (self.folder / (c.P + 'action_a0.txt')).unlink(); self.assertFalse(self.audit()['passed']) def test_scaled_cannot_replace_verified_c(self): vec_file(self.folder / (c.P + 'correction_a0.txt'), [-.25, 0., 0.]); self.assertFalse(self.audit()['passed']) def test_corrupt_rhs_fails(self): vec_file(self.folder / (c.P + 'rhs_a0.txt'), [-1., 0., 0.]); self.assertFalse(self.audit()['passed']) def test_reset_budget_fails(self): self.make_case('budget'); self.mutate_row('trials', 'attempts_total', 0, 1); self.assertFalse(self.audit()['passed']) def test_deferred_gate_corruption_fails(self): self.make_case('near'); self.mutate_row('trials', 'excess_ratio', .5); self.assertFalse(self.audit()['passed']) def test_wrong_beta_fails(self): self.mutate_row('candidates', 'beta', .25, 1); self.assertFalse(self.audit()['passed']) def test_later_beta_after_acceptance_fails(self): self.mutate_row('candidates', 'accepted', 1, 0); self.assertFalse(self.audit()['passed']) def test_selected_stale_residual_fails(self): vec_file(self.folder / (c.P + 'selected_residual_a0.txt'), [2.1, 0., 0.]); self.assertFalse(self.audit()['passed']) def test_missing_physical_replay_fails(self): (self.folder / 'coupled_trial_physical.csv').unlink(); self.assertFalse(self.audit()['passed']) def test_campaign_hash_corruption_suppresses_observations(self): self.mutate_row('candidates', 'merit', 99.); result = c.inspect_campaign(self.root) self.assertFalse(result['passed']); self.assertFalse(result['cases']['synthetic']['observations_valid']) def test_positive_predicted_decrease_required(self): self.assertFalse(c.eligibility(1., 1., 1., 1.)[0]) def test_beta_remainder_identity(self): result = c.correction_mechanism([1.], [2.], [1.7], [-.9], [-1.], .5) self.assertLess(result['decomposition_error_norm'], 1.e-15) self.assertAlmostEqual(result['own_taylor_remainder_norm'], .15) if __name__ == '__main__': unittest.main()