"""Exclusive small-evidence publication after both Phase 4AJ seals verify. Publication verifies and copies bytes. It never runs a brain, re-evaluates a scientific gate, converts FAIL/INCOMPLETE into PASS, or changes frozen sources. The source-pinned Phase 4AI verification/copy helpers are executed unchanged. """ from __future__ import annotations import hashlib import importlib.util import json from pathlib import Path, PurePosixPath ROOT = Path('/data/digital-drosophila') MODEL_RELATIVE = Path('models/male-cns-lif/apl-timing-lab-v1') EXPERIMENT_RELATIVE = Path('experiments/phase4aj-apl-timing') REGISTRY_RELATIVE = Path('registry/phase4aj-apl-timing') HELPER_RELATIVE = Path('models/male-cns-lif/pn-recruitment-audit-v1/publication.py') HELPER_SHA256 = '7a79c499098b74163e7bb16adabcbc54e7069dafa204d0b99c7f4c03d79f8016' MAX_EVIDENCE_BYTES = 4 * 1024 ** 2 EVIDENCE_SUFFIXES = frozenset(('.json', '.csv', '.md', '.log', '.sha256')) FINAL_STATUSES = frozenset(( 'PASS-FIXED-CAL-APL-TIMING-HYPOTHESIS', 'FAIL-FIXED-CAL-APL-TIMING-HYPOTHESIS', 'FAIL-FROZEN-INTEGRITY-OR-REGRESSION', 'FAIL-NUMERICAL-OR-CONTROL-GATE', 'INCOMPLETE-NUMERICAL-OR-CAL-EXECUTION', )) REQUIRED_EVIDENCE = ('audit/final-status.json', 'regression/frozen-five.json', 'regression/old-integrity-after.json') def require(condition, message): if not condition: raise ValueError(message) def load_helpers(path): """Instantiate the unchanged module from its exact pinned source bytes. Using module_from_spec plus the source bytes avoids writing/importing a pyc inside the sealed old model, even if the publisher was invoked without -B. No old module global is redirected or modified after module initialization. """ path = Path(path) require(path.is_absolute() and path.is_file() and not path.is_symlink(), 'Invalid frozen helper path') data = path.read_bytes() require(hashlib.sha256(data).hexdigest() == HELPER_SHA256, 'Frozen AI publication helper hash mismatch') spec = importlib.util.spec_from_file_location('phase4aj_frozen_ai_publication_helpers', path) require(spec is not None and spec.loader is not None, 'Unable to define frozen helper module') module = importlib.util.module_from_spec(spec) exec(compile(data, str(path), 'exec'), module.__dict__) require(module._digest(path) == HELPER_SHA256, 'Helper source changed while loading') return module def _copy(helper, source, destination, size, checksum): """Separate local delegation point for interruption-only synthetic tests.""" helper._copy_verified(source, destination, size, checksum) def _write_small_json(helper, path, value): data = (json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + '\n').encode('utf-8') require(len(data) <= MAX_EVIDENCE_BYTES, 'Generated registry evidence exceeds 4 MiB: ' + str(path)) with Path(path).open('xb') as stream: stream.write(data) require(Path(path).stat().st_size == len(data), 'Generated registry evidence size changed') def _status_evidence(helper, experiment, rows): for relative in REQUIRED_EVIDENCE: require(relative in rows, 'Required finalized evidence absent from seal: ' + relative) require(rows[relative]['bytes'] <= MAX_EVIDENCE_BYTES, 'Required finalized evidence exceeds 4 MiB') status, reference, integrity = [helper._json_bytes((experiment / relative).read_bytes()) for relative in REQUIRED_EVIDENCE] require(isinstance(status, dict) and status.get('status') in FINAL_STATUSES, 'Unrecognized Phase 4AJ final status') for key in ('numerical_gate_pass', 'full_CAL_execution_gate_pass', 'delay0_representation_pass', 'delay18_representation_pass', 'old_frozen_integrity_and_five_reference_pass'): require(type(status.get(key)) is bool, 'Missing/non-Boolean finalized result: ' + key) for key in ('candidate_generated', 'learning_achieved', 'heldout_generalization_tested', 'biological_delay_measured', 'whole_brain_equal_inhibition_integral_claim', 'universal_equivalence_claim'): require(status.get(key) is False, 'Unexpected finalized scope claim: ' + key) require(status.get('shared_APL_arithmetic_oracle_limitation') is True, 'Missing shared-APL-component oracle limitation') require(isinstance(reference, dict) and type(reference.get('pass_all')) is bool and isinstance(reference.get('trials'), list), 'Missing/invalid frozen reference report') require(isinstance(integrity, dict) and bool(integrity) and all( isinstance(row, dict) and type(row.get('pass')) is bool for row in integrity.values()), 'Missing/invalid frozen-integrity report') # This is consistency of declarations, not a new scientific evaluation. frozen_declared = reference['pass_all'] and all(row['pass'] for row in integrity.values()) require(status['old_frozen_integrity_and_five_reference_pass'] == frozen_declared, 'Final status contradicts its frozen-integrity/reference reports') if status['status'].startswith('PASS-'): require(status['numerical_gate_pass'] and status['full_CAL_execution_gate_pass'] and status['delay18_representation_pass'] and frozen_declared, 'Declared PASS contradicts finalized gate flags') return status, reference, integrity def publish(root=ROOT, helper_path=None): """Publish the fixed AJ registry exclusively; root override is for tests. Only sealed small experiment evidence and the two model seal controls are copied. Every bulk payload is nevertheless verified before and after copying. An interrupted registry is preserved and cannot be overwritten or resumed. """ root = Path(root) require(root.is_absolute() and root.is_dir() and not root.is_symlink(), 'Invalid project root') root = root.resolve(strict=True) model, experiment, registry = (root / part for part in (MODEL_RELATIVE, EXPERIMENT_RELATIVE, REGISTRY_RELATIVE)) require(not registry.exists() and not registry.is_symlink(), 'Registry already exists; no overwrite/resume') helper_path = root / HELPER_RELATIVE if helper_path is None else Path(helper_path) helper = load_helpers(helper_path) for path in (model, experiment, registry.parent): current = root for part in path.relative_to(root).parts: current = current / part require(not current.is_symlink(), 'Symlink in publication root path: ' + str(current)) verified = {'model': helper.verify_seal(model), 'experiment': helper.verify_seal(experiment)} experiment_rows = {row['path']: row for row in verified['experiment']['files']} status, reference, integrity = _status_evidence(helper, experiment, experiment_rows) plans, omitted = [], [] for row in verified['experiment']['files'] + helper._control_rows(experiment, verified['experiment']): if PurePosixPath(row['path']).suffix.lower() in EVIDENCE_SUFFIXES and row['bytes'] <= MAX_EVIDENCE_BYTES: plans.append((experiment, row, 'evidence/' + row['path'])) else: omitted.append(dict(path=row['path'], bytes=row['bytes'], sha256=row['sha256'], reason='exceeds_4_MiB' if row['bytes'] > MAX_EVIDENCE_BYTES else 'bulk_or_non_evidence_extension')) for row in helper._control_rows(model, verified['model']): require(row['bytes'] <= MAX_EVIDENCE_BYTES, 'Model seal control exceeds small-evidence limit') plans.append((model, row, 'model/' + row['path'])) destinations = {target for _, _, target in plans} for relative in ('manifest.json', 'checksums.sha256', *REQUIRED_EVIDENCE): require('evidence/' + relative in destinations, 'Required experiment evidence cannot be copied: ' + relative) require(len(destinations) == len(plans), 'Duplicate planned evidence destination') registry.parent.mkdir(parents=True, exist_ok=True) registry.mkdir(exist_ok=False) try: records = [] for source_root, row, relative in plans: source = helper._regular_file(source_root.resolve(), row['path']) _copy(helper, source, registry / relative, row['bytes'], row['sha256']) records.append(dict(source=source.relative_to(root).as_posix(), copy=relative, bytes=row['bytes'], sha256=row['sha256'])) after = {'model': helper.verify_seal(model), 'experiment': helper.verify_seal(experiment)} require(after == verified, 'A sealed tree changed during publication') require(helper._digest(helper_path) == HELPER_SHA256, 'Frozen helper changed during publication') proof = {name: {key: value for key, value in report.items() if key != 'files'} for name, report in verified.items()} _write_small_json(helper, registry / 'seal-verification.json', proof) _write_small_json(helper, registry / 'evidence-index.json', records) files = [dict(path=path.relative_to(registry).as_posix(), bytes=path.stat().st_size, sha256=helper._digest(path)) for path in sorted(registry.rglob('*')) if path.is_file()] receipt = dict(publication_complete=True, registry=registry.relative_to(root).as_posix(), project_root=str(root), declared_audit_status=status['status'], declared_gate_results={key: status[key] for key in ('numerical_gate_pass', 'full_CAL_execution_gate_pass', 'delay0_representation_pass', 'delay18_representation_pass', 'old_frozen_integrity_and_five_reference_pass')}, frozen_reference_report=dict(path='evidence/regression/frozen-five.json', sha256=experiment_rows['regression/frozen-five.json']['sha256'], declared_pass_all=reference['pass_all'], reported_trial_count=len(reference['trials'])), frozen_integrity_report=dict(path='evidence/regression/old-integrity-after.json', sha256=experiment_rows['regression/old-integrity-after.json']['sha256'], declared_pass_by_root={name: row['pass'] for name, row in integrity.items()}), frozen_helper_source=str(helper_path), frozen_helper_sha256=HELPER_SHA256, publisher_sha256=helper._digest(Path(__file__)), helper_globals_redirected=False, model_manifest_sha256=verified['model']['manifest_sha256'], experiment_manifest_sha256=verified['experiment']['manifest_sha256'], source_trees_verified_before_and_after_copy=True, copied_evidence_files=len(records), copied_evidence_bytes=sum(row['bytes'] for row in records), files=files, experiment_payload_not_copied=omitted, model_payload_not_duplicated=True, model_payload_remains_at=MODEL_RELATIVE.as_posix(), bulk_remains_at=EXPERIMENT_RELATIVE.as_posix(), complete_bulk_backup=False, scientific_gate_executed_by_publisher=False, scientific_gate_certified_by_publication=False, publication_success_is_not_scientific_pass=True, candidate_generated=False, learning_achieved=False, existing_files_modified=False, receipt_self_hash='Not self-referential; returned by publish() and printed by main()', limits=dict(max_selected_evidence_file_bytes=MAX_EVIDENCE_BYTES, selected_suffixes=sorted(EVIDENCE_SUFFIXES))) receipt_path = registry / 'publication-receipt.json' _write_small_json(helper, receipt_path, receipt) return dict(publication_complete=True, registry=str(registry), declared_audit_status=status['status'], copied_evidence_files=len(records), receipt_sha256=helper._digest(receipt_path), scientific_gate_executed_by_publisher=False, publication_success_is_not_scientific_pass=True) except Exception as error: failure = registry / 'publication-failure.json' if not failure.exists(): _write_small_json(helper, failure, dict(publication_complete=False, type=type(error).__name__, message=str(error), partial_outputs_preserved=True, existing_files_modified=False)) raise def main(): print(json.dumps(publish(), ensure_ascii=False, allow_nan=False), flush=True) if __name__ == '__main__': main()