"""Synthetic sealed documents only; no real run, registry or scientific gate.""" import hashlib import json from pathlib import Path import tempfile import unittest import describe_phase4aj as descriptive import publish_phase4aj as publication HELPER = Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py' if not HELPER.is_file(): HELPER = publication.ROOT / publication.HELPER_RELATIVE def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def write(path, value): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2), encoding='utf-8') def seal(root): rows = [dict(path=path.relative_to(root).as_posix(), bytes=path.stat().st_size, sha256=digest(path)) for path in sorted(root.rglob('*')) if path.is_file() and path.name not in ('manifest.json', 'checksums.sha256')] write(root / 'manifest.json', dict(root=str(root), files=rows)) lines = [row['sha256'] + ' ' + row['path'] for row in rows] lines.append(digest(root / 'manifest.json') + ' manifest.json') (root / 'checksums.sha256').write_text('\n'.join(lines) + '\n', encoding='utf-8') def origin_trial(name, mode, seed, odor): group = lambda positive, negative: dict(events=3, positive_events=2, negative_events=1, positive_weight_SI=positive, negative_abs_weight_SI=negative) return dict(case=name, mode=mode, seed=seed, odor_id=odor, repeat=0, pass_all=True, exact={name: True for name in ('step', 'neuron_idx', 'counts', 'v', 'g', 'ref')}, first_any_KC_tick=40, earliest_tick=20, cohort_neurons=2, cohort_before_first_KC=True, cohort_relation_to_first_KC='before', pre_kc_window_censored=False, cohort_ALLN_outgoing_events=5, outgoing_events_with_since_reset_design_antecedent=3, outgoing_events_without_since_reset_design_antecedent=2, compact_join_pair_count=17, window_source_accounting={descriptive.WINDOW: {'design_ALPN': group(.25, .03125), 'non_design_ALPN': group(.125, .0625), 'ALLN': group(.0625, .125)}, 'never_active_pre_KC': {'ALLN': group(99999, 99999)}}) class DescriptionTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix='phase4aj-description-synthetic-') self.addCleanup(temporary.cleanup) self.root = Path(temporary.name).resolve() self.model = self.root / descriptive.MODEL_RELATIVE self.experiment = self.root / descriptive.EXPERIMENT_RELATIVE self.registry = self.root / descriptive.REGISTRY_RELATIVE self.model.mkdir(parents=True) self.experiment.mkdir(parents=True) for name in ('al_origin_audit.py', 'calibration.py'): (self.model / name).write_text('# Synthetic producer placeholder, never executed\n', encoding='utf-8') self.status = dict(status='FAIL-FIXED-CAL-APL-TIMING-HYPOTHESIS', numerical_gate_pass=True, full_CAL_execution_gate_pass=True, delay0_representation_pass=False, delay18_representation_pass=False, old_frozen_integrity_and_five_reference_pass=True, candidate_generated=False, learning_achieved=False, heldout_generalization_tested=False, biological_delay_measured=False, whole_brain_equal_inhibition_integral_claim=False, universal_equivalence_claim=False, shared_APL_arithmetic_oracle_limitation=True, gates={'fixtures': {'pass_all': True}}, comparisons={'synthetic': True}, repeats={'synthetic': True}, controls={'synthetic': True}, components={'synthetic': True}) write(self.experiment / 'audit/final-status.json', self.status) write(self.experiment / 'regression/frozen-five.json', {'pass_all': True, 'trials': [{'synthetic': True}] * 5}) write(self.experiment / 'regression/old-integrity-after.json', {'synthetic-frozen-tree': {'pass': True}}) metric = {key: 0.125 for key in descriptive.CAL_MODE_FIELDS} metric.update(gate=False, pairs=[{'gate': False, 'synthetic': 'retained'}], conditions={'mean_sparsity': False}, pairwise_baseline='synthetic original pair baseline', extra_existing_field='must remain untouched') self.calibration = dict(pass_all=False, execution_gate_pass=True, numerical_integrity_pass=True, legacy_calibration_sha256=descriptive.LEGACY_CALIBRATION_SHA256, arms={arm: dict(original_representation_gate_pass=False, numerical_integrity_pass=True, modes={mode: dict(metric) for mode in descriptive.MODES}) for arm in descriptive.ARMS}) write(self.experiment / 'calibration/summary.json', self.calibration) inventory, receipts = [], [] for mode in descriptive.MODES: for odor in range(6): for seed in range(51101, 51106): name = f'cal-{mode}-{odor:02d}-s{seed}' inventory.append(dict(name=name, mode=mode, seed=seed, odor_id='synthetic-odor-' + str(odor))) trial = origin_trial(name, mode, seed, 'synthetic-odor-' + str(odor)) path = self.experiment / 'origin-audit' / name / 'summary.json' write(path, trial) receipts.append(dict(case=name, summary_sha256=digest(path))) write(self.experiment / 'protocol/calibration-cases.json', inventory) write(self.experiment / 'origin-audit/summary.json', dict(status='PASS-CONDITIONAL-EARLIEST-AL-ORIGIN-AUDIT', pass_all=True, unique_cause_proven=False, trials=receipts)) seal(self.model) seal(self.experiment) publication.publish(self.root, HELPER) def describe(self): return descriptive.describe(self.root, HELPER) def test_full_mode_dicts_fail_status_and_scope_are_retained(self): receipt = digest(self.registry / 'publication-receipt.json') result = self.describe() self.assertEqual(result['declared_final_status'], self.status['status']) path = self.registry / 'interpretation-metrics.json' saved = json.loads(path.read_text(encoding='utf-8')) for arm in descriptive.ARMS: self.assertEqual(saved['original_CAL_mode_summaries'][arm], self.calibration['arms'][arm]['modes']) self.assertFalse(saved['calibration_boolean_flags']['pass_all']) self.assertFalse(saved['new_scientific_gate']) self.assertFalse(saved['unique_cause_proven']) self.assertFalse(saved['learning_achieved']) self.assertFalse(saved['inherited_AI_other_population_or_window_share']) self.assertIn('control', saved['origin_scope']) self.assertEqual(saved['origin_window'], descriptive.WINDOW) self.assertEqual(digest(self.registry / 'publication-receipt.json'), receipt) self.assertEqual((self.registry / 'interpretation-metrics.sha256').read_text(), digest(path) + ' interpretation-metrics.json\n') def test_only_fixed_cohort_window_aggregated_and_counts_not_conflated(self): self.describe() result = json.loads((self.registry / 'interpretation-metrics.json').read_text(encoding='utf-8')) for value in result['origin_by_mode'].values(): self.assertEqual(value['trial_count'], 30) self.assertEqual(value['source_groups']['ALLN']['positive_weight_SI'], 30 * .0625) self.assertEqual(value['direct_ALPN_combined']['positive_weight_SI'], 30 * .375) self.assertEqual(value['source_groups']['ALLN']['negative_abs_weight_SI'], 30 * .125) self.assertEqual(value['first_before_KC_trial_count'], 30) self.assertEqual(value['cohort_ALLN_outgoing_events'], 150) self.assertEqual(value['outgoing_events_with_since_reset_design_antecedent'], 90) self.assertEqual(value['compact_join_pair_count'], 510) def test_never_overwrite_either_existing_output(self): path = self.registry / 'interpretation-metrics.sha256' path.write_bytes(b'keep partial checksum\n') with self.assertRaisesRegex(ValueError, 'already exists'): self.describe() self.assertEqual(path.read_bytes(), b'keep partial checksum\n') self.assertFalse((self.registry / 'interpretation-metrics.json').exists()) def test_published_receipt_or_sealed_input_corruption_blocks(self): path = self.experiment / 'calibration/summary.json' path.write_bytes(b'x' * path.stat().st_size) with self.assertRaisesRegex(ValueError, 'Payload SHA-256 mismatch'): self.describe() self.assertFalse((self.registry / 'interpretation-metrics.json').exists()) def test_success_cannot_be_rerun(self): self.describe() path = self.registry / 'interpretation-metrics.json' before = digest(path) with self.assertRaisesRegex(ValueError, 'already exists'): self.describe() self.assertEqual(digest(path), before) def test_censoring_and_first_tick_relation_are_not_relabelled(self): trial = origin_trial('synthetic', 'physiological', 51101, 'synthetic') trial.update(first_any_KC_tick=-1, cohort_before_first_KC=False, cohort_relation_to_first_KC='KC_unobserved', pre_kc_window_censored=True) result = descriptive.aggregate_origin([trial]) self.assertEqual(result['first_before_KC_trial_count'], 0) self.assertEqual(result['first_KC_window_censored_trial_count'], 1) self.assertEqual(result['cohort_relation_trial_counts'], {'KC_unobserved': 1}) if __name__ == '__main__': unittest.main()