"""Temporary-directory, CPU-only publication mutations; no real experiment.""" from __future__ import annotations import hashlib import json from pathlib import Path import tempfile import unittest from unittest import mock import publish_phase4ao as P _KNOWN_HELPERS = (P.ROOT / P.HELPER_REL, Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py') HELPER = next((p for p in _KNOWN_HELPERS if p.is_file()), None) if HELPER is None or P.digest(HELPER) != P.HELPER_SHA: raise ValueError('Pinned publication helper absent or changed in server/local source layout') def put(path, value): path.parent.mkdir(parents=True, exist_ok=True) data = value if isinstance(value, bytes) else (json.dumps(value, sort_keys=True) + '\n').encode() path.write_bytes(data) return hashlib.sha256(data).hexdigest() def read(path): return json.loads(path.read_text(encoding='utf-8')) def seal(exp): rows = [dict(path=p.relative_to(exp).as_posix(), bytes=p.stat().st_size, sha256=P.digest(p)) for p in sorted(exp.rglob('*')) if p.is_file() and p.relative_to(exp).as_posix() not in ('manifest.json', 'checksums.sha256')] put(exp / 'manifest.json', dict(root=str(exp), files=rows)) data = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows) data += P.digest(exp / 'manifest.json') + ' manifest.json\n' put(exp / 'checksums.sha256', data.encode()) class PublisherTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name).resolve() self.exp = self.root / P.EXP_REL self.registry = self.root / P.REGISTRY_REL self.model = self.root / P.MODEL_REL put(self.root / P.HELPER_REL, HELPER.read_bytes()) def fixture(self, status='PASS-EXACT-TWO-EPOCH-DELTA-INTERFACE-NOT-LEARNING', count=3): sources, prepared = {}, {} names = ['worker.py', 'test_worker.py', 'PROTOCOL.md'] + [f'source_{i:03d}.py' for i in range(max(0, count - 3))] for name in names[:count]: raw = ('# synthetic registered source ' + name + '\n').encode() original = self.model / name snapshot = self.exp / P.SNAPSHOT / name sha = put(original, raw) put(snapshot, raw) sources[str(original)] = sha prepared[str(snapshot)] = sha full_plan = put(self.exp / P.FULL_PLAN, dict(draft_only=True, no_brain_inputs=True)) prepared[str(self.exp / P.FULL_PLAN)] = full_plan self.registration = dict(schema='phase4ao-conditioning-interface-registration-v1', source_sha256=sources, prepared_file_sha256=prepared, full_trial_plan_file_sha256=full_plan, git_code_commit='4' * 40, committed_source_blobs_verified=len(sources)) self.reg_sha = put(self.exp / P.REGISTRATION, self.registration) analysis_path = P.FAILURE_ANALYSIS if status.startswith('FAIL') else P.ANALYSIS analysis_sha = put(self.exp / analysis_path, dict(status=status, registration_sha256=self.reg_sha, complete_conditioning_executed=False, completed_rows=[] if status.startswith('FAIL') else ['synthetic'])) self.final = dict(status=status, registration_sha256=self.reg_sha, analysis_sha256=analysis_sha, learned_candidate=False, learning_success_claimed=False, full_conditioning_executed=False) put(self.exp / P.FINAL, self.final) put(self.exp / 'logs/worker.log', b'partial failures are retained\n') put(self.exp / 'runs/first/summary.json', dict(complete=True)) put(self.exp / 'runs/first/state.npz', b'fake binary') put(self.exp / 'runs/first/hashes.jsonl', b'{}\n') put(self.exp / 'extension-build/compiler.log', b'build output not exported') put(self.exp / 'reference.npy', b'not a text file') put(self.exp / 'large.json', b' ' * (P.MAX_BYTES + 1)) self.reseal() def reseal(self): seal(self.exp) self.anchors = (P.digest(self.exp / P.REGISTRATION), P.digest(self.exp / 'manifest.json'), P.digest(self.exp / P.FINAL)) def publish(self, *anchors): return P.publish(*(anchors or self.anchors), root=self.root) def test_success_complete_small_seal_inventory_and_51_dynamic_sources(self): self.fixture(count=51) original = {p: P.digest(p) for p in self.exp.rglob('*') if p.is_file()} result = self.publish() self.assertTrue(result['publication_complete']) self.assertEqual(result['source_count'], 51) self.assertFalse(result['complete_bulk_backup']) helper = P.load_helper(self.root / P.HELPER_REL) verified = helper.verify_seal(self.registry) self.assertEqual(verified['manifest_sha256'], result['registry_manifest_sha256']) receipt = read(self.registry / 'publication-receipt.json') self.assertEqual(result['receipt_sha256'], P.digest(self.registry / 'publication-receipt.json')) self.assertEqual(receipt['registration_sha256'], self.anchors[0]) self.assertFalse(receipt['scientific_gate_certified_by_publication']) for entry in receipt['files']: self.assertEqual(P.digest(self.registry / entry['path']), entry['sha256']) inventory = read(self.registry / 'artifact-inventory.json')['files'] rows = {r['path']: r for r in inventory} self.assertEqual(set(rows), {p.relative_to(self.exp).as_posix() for p in original}) self.assertEqual(rows[P.FULL_PLAN]['reason'], 'full_trial_plan_intentionally_not_exported') self.assertEqual(rows['large.json']['reason'], 'exceeds_4_MiB') for name in ('runs/first/state.npz', 'reference.npy', 'runs/first/hashes.jsonl', 'extension-build/compiler.log'): self.assertEqual(rows[name]['disposition'], 'omitted') self.assertEqual(rows[name]['sha256'], P.digest(self.exp / name)) self.assertEqual((self.registry / 'evidence/manifest.json').read_bytes(), (self.exp / 'manifest.json').read_bytes()) self.assertEqual((self.registry / 'evidence/checksums.sha256').read_bytes(), (self.exp / 'checksums.sha256').read_bytes()) self.assertTrue(all(p.stat().st_size <= P.MAX_BYTES for p in self.registry.rglob('*') if p.is_file())) self.assertEqual(original, {p: P.digest(p) for p in self.exp.rglob('*') if p.is_file()}) def test_fail_with_partial_run_no_execution_complete_is_preserved(self): self.fixture('FAIL-POSTPROCESS-MODULE-CACHE') result = self.publish() self.assertEqual(result['declared_audit_status'], 'FAIL-POSTPROCESS-MODULE-CACHE') receipt = read(self.registry / 'publication-receipt.json') self.assertEqual(receipt['analysis_path'], P.FAILURE_ANALYSIS) self.assertFalse((self.registry / 'evidence/audit/execution-complete.json').exists()) self.assertFalse(receipt['learning_achieved']) def test_incomplete_status_is_not_upgraded(self): self.fixture('INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE') self.assertEqual(self.publish()['declared_audit_status'], 'INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE') def test_three_external_anchors_required_and_checked(self): self.fixture() for index in range(3): values = list(self.anchors) values[index] = '0' * 64 with self.subTest(index=index), self.assertRaisesRegex(ValueError, 'External anchor mismatch'): self.publish(*values) with self.assertRaises(ValueError): self.publish('', self.anchors[1], self.anchors[2]) self.assertFalse(self.registry.exists()) def test_existing_registry_is_never_touched(self): self.fixture() marker = self.registry / 'do-not-touch.txt' put(marker, b'keep') with self.assertRaisesRegex(ValueError, 'already exists'): self.publish() self.assertEqual(marker.read_bytes(), b'keep') self.assertEqual(len(list(self.registry.iterdir())), 1) def test_original_source_mutation_rejected_even_with_valid_experiment_seal(self): self.fixture() put(self.model / 'worker.py', b'# other source') with self.assertRaisesRegex(ValueError, 'differ'): self.publish() self.assertFalse(self.registry.exists()) def test_snapshot_mutation_resealed_still_rejected_by_registration(self): self.fixture() put(self.exp / P.SNAPSHOT / 'worker.py', b'# changed') self.reseal() with self.assertRaisesRegex(ValueError, 'differ'): self.publish() def test_extra_snapshot_is_not_silently_exported(self): self.fixture() put(self.exp / P.SNAPSHOT / 'unregistered.py', b'# not registered') self.reseal() with self.assertRaisesRegex(ValueError, 'snapshot inventory'): self.publish() def test_source_escape_rejected_after_self_consistent_resealing(self): self.fixture() old = str(self.model / 'worker.py') checksum = self.registration['source_sha256'].pop(old) external = self.root / 'unrelated/worker.py' put(external, (self.model / 'worker.py').read_bytes()) self.registration['source_sha256'][str(external)] = checksum self.reg_sha = put(self.exp / P.REGISTRATION, self.registration) analysis = read(self.exp / P.ANALYSIS) analysis['registration_sha256'] = self.reg_sha self.final['analysis_sha256'] = put(self.exp / P.ANALYSIS, analysis) self.final['registration_sha256'] = self.reg_sha put(self.exp / P.FINAL, self.final) self.reseal() with self.assertRaisesRegex(ValueError, 'escapes the fixed model'): self.publish() def test_unlisted_file_or_bad_checksum_control_rejected(self): self.fixture() put(self.exp / 'unlisted.txt', b'not sealed') with self.assertRaisesRegex(ValueError, 'unlisted'): self.publish() (self.exp / 'unlisted.txt').unlink() with (self.exp / 'checksums.sha256').open('ab') as handle: handle.write(b'bad line\n') with self.assertRaisesRegex(ValueError, 'Malformed checksums'): self.publish() def test_final_analysis_link_and_status_must_agree(self): self.fixture() self.final['analysis_sha256'] = '0' * 64 put(self.exp / P.FINAL, self.final) self.reseal() with self.assertRaisesRegex(ValueError, 'analysis link'): self.publish() self.final['analysis_sha256'] = P.digest(self.exp / P.ANALYSIS) self.final['status'] = 'FAIL-OTHER' put(self.exp / P.FINAL, self.final) self.reseal() with self.assertRaisesRegex(ValueError, 'identities differ'): self.publish() def test_no_learning_claim_and_helper_pin_cannot_be_bypassed(self): self.fixture() self.final['learning_success_claimed'] = True put(self.exp / P.FINAL, self.final) self.reseal() with self.assertRaisesRegex(ValueError, 'scope flag'): self.publish() put(self.root / P.HELPER_REL, b'# alternate helper') with self.assertRaisesRegex(ValueError, 'helper SHA mismatch'): self.publish() def test_postcopy_full_seal_detects_change_in_omitted_bulk_and_preserves_partial(self): self.fixture() helper = P.load_helper(self.root / P.HELPER_REL) original_copy = helper._copy_verified changed = False def interrupted_copy(*args): nonlocal changed original_copy(*args) if not changed: changed = True put(self.exp / 'reference.npy', b'changed omitted bytes') # Test-local injected failure only; production never changes helper globals. with mock.patch.object(helper, '_copy_verified', side_effect=interrupted_copy), mock.patch.object(P, 'load_helper', return_value=helper): with self.assertRaisesRegex(ValueError, 'Payload byte count|Payload SHA'): self.publish() self.assertTrue(self.registry.exists()) self.assertTrue(read(self.registry / 'publication-failure.json')['partial_outputs_preserved']) self.assertFalse((self.registry / 'publication-receipt.json').exists()) with self.assertRaisesRegex(ValueError, 'already exists'): self.publish() def test_small_registry_seal_catches_copied_payload_mutation(self): self.fixture() self.publish() path = self.registry / 'evidence/logs/worker.log' put(path, b'tampered after publication') helper = P.load_helper(self.root / P.HELPER_REL) with self.assertRaisesRegex(ValueError, 'Payload byte count|Payload SHA'): helper.verify_seal(self.registry) def test_valid_reseal_between_anchor_read_and_verification_is_rejected(self): self.fixture() helper = P.load_helper(self.root / P.HELPER_REL) original_verify = helper.verify_seal def resealed_verify(directory): put(self.exp / 'logs/worker.log', b'changed after external anchor read') seal(self.exp) return original_verify(directory) with mock.patch.object(helper, 'verify_seal', side_effect=resealed_verify), mock.patch.object(P, 'load_helper', return_value=helper): with self.assertRaisesRegex(ValueError, 'differs from external'): self.publish() self.assertFalse(self.registry.exists()) if __name__ == '__main__': unittest.main()