From a9c59184457eb729970ba7965b44707a84fc142b Mon Sep 17 00:00:00 2001 From: Digital Drosophila Archive Date: Wed, 30 Sep 2026 19:59:56 +0800 Subject: [PATCH] Add exclusive evidence publication for execution audit --- .../phase4aq-publication/publish_phase4aq.py | 296 ++++++++++++++++++ .../test_publish_phase4aq.py | 193 ++++++++++++ 2 files changed, 489 insertions(+) create mode 100644 tools/phase4aq-publication/publish_phase4aq.py create mode 100644 tools/phase4aq-publication/test_publish_phase4aq.py diff --git a/tools/phase4aq-publication/publish_phase4aq.py b/tools/phase4aq-publication/publish_phase4aq.py new file mode 100644 index 0000000..86dd592 --- /dev/null +++ b/tools/phase4aq-publication/publish_phase4aq.py @@ -0,0 +1,296 @@ +"""Publish only small evidence from a fully sealed normal AQ engineering run. + +Three external SHA-256 anchors are mandatory. No simulation, numerical audit, +learning gate or historical integrity run is executed here. FAIL/INCOMPLETE +remain FAIL/INCOMPLETE. Large payloads are verified but never copied; this is +not a bulk backup. Only a new, absent AQ registry is created. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path, PurePosixPath +import re +import sys +from types import ModuleType + +ROOT = Path('/data/digital-drosophila') +MODEL_REL = Path('models/male-cns-lif/conditioning-execution-v1') +EXP_REL = Path('experiments/phase4aq-conditioning-execution') +REGISTRY_REL = Path('registry/phase4aq-conditioning-execution') +HELPER_REL = Path('models/male-cns-lif/pn-recruitment-audit-v1/publication.py') +HELPER_SHA = '7a79c499098b74163e7bb16adabcbc54e7069dafa204d0b99c7f4c03d79f8016' +REGISTRATION, FINAL = 'protocol/preregistered.json', 'audit/final-status.json' +SNAPSHOT = 'protocol/source-snapshot/' +REPORTS = dict(numerical_sha256='audit/numerical-comparison.json', budget_sha256='audit/budget-report.json', + integrity_after_sha256='regression/integrity-after.json', integrity_at_seal_sha256='regression/integrity-at-seal.json') +STATUSES = frozenset(('PASS-EXACT-UPDATED-BOUNDARY-ENGINEERING-NOT-LEARNING', + 'FAIL-NUMERICAL-OR-LINEAGE', 'INCOMPLETE-CHANGED-EDGE-COVERAGE')) +MAX_BYTES = 4 * 1024**2 +SUFFIXES = frozenset(('.json', '.md', '.py', '.log', '.sha256')) +SHA = re.compile('[0-9a-f]{64}') + + +def require(ok, message): + if not ok: + raise ValueError(message) + + +def digest(path): + h = hashlib.sha256() + with Path(path).open('rb') as stream: + for chunk in iter(lambda: stream.read(8 << 20), b''): + h.update(chunk) + return h.hexdigest() + + +def load_helper(path): + raw = Path(path).read_bytes() + require(hashlib.sha256(raw).hexdigest() == HELPER_SHA, 'Frozen helper SHA mismatch') + module = ModuleType('_phase4aq_publication_pure_seal_helper') + module.__file__ = str(path) + exec(compile(raw, str(path), 'exec'), module.__dict__) + return module + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False) + + +def write_json(path, value): + raw = (json.dumps(value, sort_keys=True, indent=2, ensure_ascii=False, allow_nan=False) + '\n').encode('utf-8') + require(len(raw) <= MAX_BYTES, 'Generated evidence exceeds 4 MiB') + with Path(path).open('xb') as stream: + stream.write(raw) + + +def small_json(helper, experiment, rows, name): + require(name in rows and rows[name]['bytes'] <= MAX_BYTES, 'Required small evidence missing or oversize: ' + name) + path = helper._regular_file(experiment, name) + require(digest(path) == rows[name]['sha256'], 'Evidence changed: ' + name) + result = helper._json_bytes(path.read_bytes()) + require(type(result) is dict, 'Expected JSON object: ' + name) + return result + + +def source_identity(helper, root, experiment, rows, reg, reg_sha): + """Actual registered dynamic inventory, not a guessed source-file count.""" + model = root / MODEL_REL; sources = reg.get('source_sha256'); prepared = reg.get('prepared_file_sha256') + require(type(sources) is dict and bool(sources) and type(prepared) is dict, 'Missing registered source/snapshot inventory') + proof = []; expected = set() + for text, sha in sorted(sources.items()): + path = Path(text) + require(type(text) is str and path.is_absolute() and path.parent == model and + path.suffix in ('.py', '.md') and '..' not in path.parts and type(sha) is str and SHA.fullmatch(sha), + 'Registered source escapes exact AQ model or has invalid identity') + rel = SNAPSHOT + path.name; expected.add(rel) + current = helper._regular_file(root, (MODEL_REL / path.name).as_posix()) + snapshot = helper._regular_file(experiment, rel) + require(rel in rows and rows[rel]['bytes'] <= MAX_BYTES and rows[rel]['sha256'] == sha and + prepared.get(str(snapshot)) == sha and current.stat().st_size == rows[rel]['bytes'] and + digest(current) == digest(snapshot) == sha, 'Registered source/snapshot/current bytes differ: ' + path.name) + proof.append(dict(model_source=(MODEL_REL / path.name).as_posix(), sealed_source=rel, + copy='source-snapshot/' + path.name, bytes=rows[rel]['bytes'], sha256=sha)) + require({n for n in rows if n.startswith(SNAPSHOT)} == expected, 'Sealed snapshot inventory differs from registration') + for text, sha in prepared.items(): + path = Path(text) + require(type(text) is str and path.is_absolute() and path.is_relative_to(experiment) and '..' not in path.parts and + type(sha) is str and SHA.fullmatch(sha), 'Invalid prepared path or SHA') + rel = path.relative_to(experiment).as_posix(); helper._relative_name(rel) + require(rel in rows and rows[rel]['sha256'] == sha, 'Prepared evidence differs from full seal') + commit = reg.get('source_code_commit') + require(type(commit) is str and re.fullmatch('[0-9a-f]{40}', commit), 'Missing source commit identity') + return dict(registration_sha256=reg_sha, source_count=len(proof), source_code_commit=commit, files=proof, + registration_current_model_sealed_snapshot_exact=True, entire_model_directory_sealed=False, + Git_blobs_reverified_by_publisher=False) + + +def declared_scope(reg, final, reports, execution, reg_sha): + require(reg.get('schema') == 'conditioning-execution-registration-v1' and reg.get('engineering_only') is True and + reg.get('scientific_acquisition_authorized') is False, 'Wrong AQ registration scope') + require(final.get('schema') == 'conditioning-execution-final-v1' and final.get('status') in STATUSES and + final.get('registration_sha256') == reg_sha and final.get('source_code_commit') == reg.get('source_code_commit') and + final.get('numerical_and_budget_recomputed_at_seal') is True, 'Not a normal finalized AQ engineering report') + for flag in ('scientific_acquisition_executed', 'scientific_acquisition_authorized', 'candidate_generated', + 'learning_pass_claimed', 'entire_model_directory_sealed', 'bulk_data_git_backup_claimed'): + require(final.get(flag) is False, 'Unexpected final scientific/backup claim: ' + flag) + numerical = reports['numerical_sha256']; budget = reports['budget_sha256'] + require(numerical.get('schema') == 'conditioning-execution-numerical-audit-v1' and numerical.get('complete') is True and + numerical.get('registration_sha256') == reg_sha and numerical.get('status') == final['status'] and + type(numerical.get('pass_all')) is bool and numerical['pass_all'] == final['status'].startswith('PASS-') and + numerical.get('learning_pass_claimed') is numerical.get('candidate_generated') is False, + 'Final/numerical status or authority differs') + require(budget.get('schema') == 'conditioning-actual-budget-report-v1' and budget.get('registration_sha256') == reg_sha and + budget.get('learning_pass_claimed') is False and budget.get('engineering_scope') is True and + budget.get('normal_budget_authorized') is False and budget.get('execution_authorized') is False, + 'Wrong budget scope; engineering publication cannot authorize training') + before, at_seal = reports['integrity_after_sha256'], reports['integrity_at_seal_sha256'] + require(canonical(before) == canonical(at_seal) and before.get('complete') is True and + before.get('AP', {}).get('pass_all') is True and type(before.get('prior')) is dict and + type(before.get('anchors')) is dict and final.get('historical_integrity_after_pass') is True, + 'Final linked independent integrity reports disagree/incomplete') + for flag in ('old_experiments_reexecuted', 'old_files_modified', 'five_frozen_reference_trials_rerun_here', + 'scientific_learning_pass_claimed'): + require(before.get(flag) is False, 'Wrong historical evidence interpretation') + require(final.get('historical_five_trial_regression') == 'AM sealed evidence preserved; not rerun here', + 'Historical regression must not be called a new run') + plan = reg.get('plan', {}); calls = plan.get('calls'); counts = plan.get('counts', {}) + require(type(calls) is list and len(calls) == 23 and all(type(c) is dict and type(c.get('call_id')) is str and + type(c.get('steps')) is int and c['steps'] > 0 for c in calls) and + len({c['call_id'] for c in calls}) == 23 and sum(c['steps'] for c in calls) == 125200, + 'Registered actual call/tick coverage differs') + required_counts = dict(new_brain_calls=23, new_brain_ticks=125200, cpu_children=7, gpu_sampler_children=1, total_children=31) + require(all(type(counts.get(k)) is int and counts[k] == v for k, v in required_counts.items()), 'Registered process counts differ') + require(execution.get('complete') is True and execution.get('registration_sha256') == reg_sha and + execution.get('completed_calls') == [c['call_id'] for c in calls] and + type(execution.get('processes')) is dict and len(execution['processes']) == 31 and + execution.get('scientific_learning_pass_claimed') is execution.get('candidate_generated') is False, + 'Normal finalizer needs a complete recorded execution inventory') + for final_key, number, numerical_key in ( + ('actual_new_brain_calls', 23, 'actual_new_brain_calls'), + ('actual_new_brain_ticks', 125200, 'actual_new_brain_ticks'), + ('actual_execution_child_processes', 31, 'actual_processes')): + require(type(final.get(final_key)) is int and type(numerical.get(numerical_key)) is int and + final[final_key] == numerical[numerical_key] == number, 'Declared final coverage differs') + require(numerical.get('actual_CPU_processes') == 7 and numerical.get('actual_GPU_sampler_processes') == 1, + 'CPU versus sampler versus brain distinction lost') + return dict(declared_engineering_status=final['status'], declared_budget_status=budget.get('status'), + declared_new_brain_calls=23, declared_new_brain_ticks=125200, declared_CPU_processes=7, + declared_GPU_sampler_processes=1, declared_total_child_processes=31, + source_scope='Declared report/registration consistency plus full sealed byte identities; numerical gate not repeated', + historical_five_trial_regression=final['historical_five_trial_regression'], + original_AO_failure_not_reclassified=True, imported_or_prior_bulk_reread_by_publisher=False, + scientific_gate_recomputed=False, learning_achieved=False, candidate_generated=False) + + +def process_references(experiment, rows, numerical, execution): + """Bind stored audit references to sealed process/job/worker bytes, not exits.""" + evidence = numerical.get('process_evidence') + require(type(evidence) is dict and len(evidence) == 31 and set(evidence) == set(execution['processes']), + 'Numerical process evidence coverage differs') + for name, value in evidence.items(): + require(type(name) is str and re.fullmatch('[A-Za-z0-9_-]+', name), 'Unsafe process name') + require(execution['processes'][name] == str(experiment / 'receipts' / (name + '.process.json')) and + type(value.get('recorded_exit_code')) is int and value['recorded_exit_code'] == 0, + 'Declared complete process identity differs') + for field, relative in (('process_sha256', 'receipts/' + name + '.process.json'), + ('worker_sha256', 'receipts/' + name + '.worker.json'), + ('job_sha256', 'jobs/' + name + '.json')): + require(relative in rows and rows[relative]['sha256'] == value.get(field), 'Numerical process file reference differs: ' + relative) + + +def selection(row): + path = PurePosixPath(row['path']) + if path.name == 'full-trial-plan.json': return None, 'full_trial_plan_not_exported' + if any(p in ('extension-build', 'build', '__pycache__') for p in path.parts): return None, 'build_or_cache_artifact' + if row['bytes'] > MAX_BYTES: return None, 'exceeds_4_MiB' + if row['path'].startswith(SNAPSHOT): return 'source-snapshot/' + path.name, None + if path.suffix.lower() in SUFFIXES: return 'evidence/' + row['path'], None + return None, 'binary_or_non_selected_extension' + + +def file_rows(root): + return [dict(path=p.relative_to(root).as_posix(), bytes=p.stat().st_size, sha256=digest(p)) + for p in sorted(root.rglob('*')) if p.is_file()] + + +def seal_registry(helper, registry): + rows = file_rows(registry) + write_json(registry / 'manifest.json', dict(root=str(registry), files=rows)) + raw = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows) + raw += digest(registry / 'manifest.json') + ' manifest.json\n' + require(len(raw.encode()) <= MAX_BYTES, 'Registry checksums exceed small evidence limit') + with (registry / 'checksums.sha256').open('xb') as stream: stream.write(raw.encode('utf-8')) + return helper.verify_seal(registry) + + +def publish(registration_sha256, experiment_manifest_sha256, final_sha256, *, root=ROOT): + require(sys.dont_write_bytecode, 'Use Python -B') + for value in (registration_sha256, experiment_manifest_sha256, final_sha256): + require(type(value) is str and SHA.fullmatch(value), 'Three external SHA-256 anchors required') + root = Path(root) + require(root.is_absolute() and root.is_dir(), 'Absolute existing project root required') + for p in (root, *root.parents): + require(not p.is_symlink() and not (hasattr(p, 'is_junction') and p.is_junction()), 'Linked project path forbidden') + root = root.resolve(strict=True) + experiment, registry, model, helper_path = (root / p for p in (EXP_REL, REGISTRY_REL, MODEL_REL, HELPER_REL)) + for path in (experiment, registry, model, helper_path): + cursor = root + for part in path.relative_to(root).parts: + cursor /= part + require(not cursor.is_symlink() and not (hasattr(cursor, 'is_junction') and cursor.is_junction()), 'Linked publication path') + require(not registry.exists(), 'Registry exists; never overwrite or resume') + helper = load_helper(helper_path); own_sha = digest(__file__) + for name, expected in ((REGISTRATION, registration_sha256), ('manifest.json', experiment_manifest_sha256), (FINAL, final_sha256)): + require(digest(helper._regular_file(experiment, name)) == expected, 'External anchor mismatch: ' + name) + verified = helper.verify_seal(experiment); rows = {r['path']: r for r in verified['files']} + require(verified['manifest_sha256'] == experiment_manifest_sha256 and rows.get(REGISTRATION, {}).get('sha256') == registration_sha256 and + rows.get(FINAL, {}).get('sha256') == final_sha256, 'Verified seal differs from external anchors') + reg, final = (small_json(helper, experiment, rows, name) for name in (REGISTRATION, FINAL)) + reports = {} + for field, name in REPORTS.items(): + require(name in rows and rows[name]['sha256'] == final.get(field), 'Final linked report SHA mismatch: ' + field) + reports[field] = small_json(helper, experiment, rows, name) + execution_name = 'audit/execution-complete.json' + execution = small_json(helper, experiment, rows, execution_name) + require(reports['numerical_sha256'].get('execution_complete_sha256') == rows[execution_name]['sha256'], 'Numerical execution inventory binding differs') + scope = declared_scope(reg, final, reports, execution, registration_sha256) + process_references(experiment, rows, reports['numerical_sha256'], execution) + sources = source_identity(helper, root, experiment, rows, reg, registration_sha256) + inventory = []; copies = [] + for row in verified['files'] + helper._control_rows(experiment, verified): + target, reason = selection(row) + inventory.append(dict(**row, copy=target, disposition='copied' if target else 'omitted', reason=reason)) + if target: copies.append((row, target)) + targets = [target for _, target in copies] + required = {'evidence/' + name for name in (REGISTRATION, FINAL, execution_name, *REPORTS.values(), 'manifest.json', 'checksums.sha256')} + required |= {v['copy'] for v in sources['files']} + require(len(set(targets)) == len(targets) and required <= set(targets), 'Required evidence omitted/oversize or destination collision') + registry.parent.mkdir(parents=True, exist_ok=True); registry.mkdir(exist_ok=False) + try: + index = [] + for row, target in copies: + helper._copy_verified(helper._regular_file(experiment, row['path']), registry / target, row['bytes'], row['sha256']) + index.append(dict(source=(EXP_REL / row['path']).as_posix(), copy=target, bytes=row['bytes'], sha256=row['sha256'])) + require(helper.verify_seal(experiment) == verified, 'Sealed experiment changed while publishing') + require(source_identity(helper, root, experiment, rows, reg, registration_sha256) == sources, 'Current registered source changed') + require(digest(helper_path) == HELPER_SHA and digest(__file__) == own_sha, 'Publisher/helper changed') + write_json(registry / 'evidence-index.json', index) + write_json(registry / 'source-identity.json', sources) + write_json(registry / 'scope-audit.json', scope) + write_json(registry / 'seal-verification.json', {k: v for k, v in verified.items() if k != 'files'}) + write_json(registry / 'artifact-inventory.json', dict(files=inventory, experiment_manifest_sha256=experiment_manifest_sha256, + omitted_files=sum(r['copy'] is None for r in inventory), omitted_bytes=sum(r['bytes'] for r in inventory if r['copy'] is None), + covers_entire_AQ_payload_and_controls=True, complete_bulk_backup=False)) + receipt = dict(schema='phase4aq-publication-v1', publication_complete=True, project_root=str(root), + registry=REGISTRY_REL.as_posix(), declared_audit_status=final['status'], declared_budget_status=scope['declared_budget_status'], + registration_sha256=registration_sha256, experiment_manifest_sha256=experiment_manifest_sha256, + experiment_checksums_sha256=verified['checksums_sha256'], final_sha256=final_sha256, + linked_reports={field: dict(path=path, sha256=final[field]) for field, path in REPORTS.items()}, + source_count=sources['source_count'], source_code_commit=sources['source_code_commit'], + publisher_sha256=own_sha, helper_sha256=HELPER_SHA, helper_globals_modified=False, + files=file_rows(registry), copied_evidence_files=len(index), copied_evidence_bytes=sum(r['bytes'] for r in index), + bulk_remains_at=EXP_REL.as_posix(), prior_imported_bulk_copied=False, complete_bulk_backup=False, + entire_model_directory_sealed=False, existing_files_modified=False, scientific_gate_executed_by_publisher=False, + scientific_gate_certified_by_publication=False, learning_achieved=False, candidate_generated=False, new_brain_runs=0, + normal_training_or_budget_authorized=False, receipt_self_hash='Returned externally and covered by registry seal; not self-referential.', + registry_controls='Written after receipt; verified independently.', limits=dict(max_selected_file_bytes=MAX_BYTES, suffixes=sorted(SUFFIXES))) + write_json(registry / 'publication-receipt.json', receipt) + sealed = seal_registry(helper, registry) + return dict(publication_complete=True, registry=str(registry), declared_audit_status=final['status'], + source_count=sources['source_count'], receipt_sha256=digest(registry / 'publication-receipt.json'), + registry_manifest_sha256=sealed['manifest_sha256'], registry_checksums_sha256=sealed['checksums_sha256'], + copied_evidence_files=len(index), complete_bulk_backup=False, scientific_gate_certified_by_publication=False) + except Exception as error: + if not (registry / 'publication-failure.json').exists(): + write_json(registry / 'publication-failure.json', dict(publication_complete=False, error_type=type(error).__name__, + error=str(error), partial_outputs_preserved=True, existing_files_modified=False)) + raise + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + for name in ('registration', 'experiment-manifest', 'final'): parser.add_argument('--' + name + '-sha256', required=True) + args = parser.parse_args() + print(json.dumps(publish(args.registration_sha256, args.experiment_manifest_sha256, args.final_sha256), sort_keys=True), flush=True) diff --git a/tools/phase4aq-publication/test_publish_phase4aq.py b/tools/phase4aq-publication/test_publish_phase4aq.py new file mode 100644 index 0000000..6f12755 --- /dev/null +++ b/tools/phase4aq-publication/test_publish_phase4aq.py @@ -0,0 +1,193 @@ +"""Pure temporary sealed trees; no real AQ publication or scientific run.""" +import copy +import json +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +import publish_phase4aq as P + +HELPER_CANDIDATES = (Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py', + Path(__file__).resolve().parents[2] / P.HELPER_REL) +HELPER = next((p for p in HELPER_CANDIDATES if p.is_file()), HELPER_CANDIDATES[0]) + + +def put(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + raw = value if type(value) is bytes else (json.dumps(value, sort_keys=True) + '\n').encode() + path.write_bytes(raw) + return P.digest(path) + + +def read(path): + return json.loads(path.read_text(encoding='utf-8')) + + +def seal(root): + rows = [r for r in P.file_rows(root) if r['path'] not in ('manifest.json', 'checksums.sha256')] + put(root / 'manifest.json', dict(root=str(root), files=rows)) + raw = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows) + put(root / 'checksums.sha256', (raw + P.digest(root / 'manifest.json') + ' manifest.json\n').encode()) + + +class PublicationTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name).resolve() + self.exp, self.model, self.registry = (self.root / x for x in (P.EXP_REL, P.MODEL_REL, P.REGISTRY_REL)) + put(self.root / P.HELPER_REL, HELPER.read_bytes()) + + def fixture(self, status='PASS-EXACT-UPDATED-BOUNDARY-ENGINEERING-NOT-LEARNING', source_count=3): + sources = {}; prepared = {} + for i in range(source_count): + name = f'new_source_{i}.py'; raw = ('# synthetic source ' + str(i)).encode() + sha = put(self.model / name, raw); sources[str(self.model / name)] = sha + prepared[str(self.exp / P.SNAPSHOT / name)] = put(self.exp / P.SNAPSHOT / name, raw) + self.reg = dict(schema='conditioning-execution-registration-v1', engineering_only=True, + scientific_acquisition_authorized=False, source_sha256=sources, prepared_file_sha256=prepared, + source_code_commit='a'*40, plan=dict(calls=[dict(call_id=f'call-{i}',steps=1000 if i<22 else 103200) for i in range(23)], + counts=dict(new_brain_calls=23,new_brain_ticks=125200,cpu_children=7,gpu_sampler_children=1,total_children=31))) + self.execution = dict(complete=True, completed_calls=[c['call_id'] for c in self.reg['plan']['calls']], + scientific_learning_pass_claimed=False,candidate_generated=False,processes={}) + evidence = {} + for i in range(31): + name = f'process-{i}'; row = {'recorded_exit_code':0} + for field, rel in (('process_sha256',f'receipts/{name}.process.json'), + ('worker_sha256',f'receipts/{name}.worker.json'),('job_sha256',f'jobs/{name}.json')): + row[field] = put(self.exp / rel, {'synthetic':name,'role':field}) + evidence[name] = row; self.execution['processes'][name] = str(self.exp / 'receipts' / (name+'.process.json')) + self.numerical = dict(schema='conditioning-execution-numerical-audit-v1',complete=True,status=status, + pass_all=status.startswith('PASS-'),learning_pass_claimed=False,candidate_generated=False, + actual_new_brain_calls=23,actual_new_brain_ticks=125200,actual_processes=31,actual_CPU_processes=7, + actual_GPU_sampler_processes=1,process_evidence=evidence) + self.budget = dict(schema='conditioning-actual-budget-report-v1', status='INSUFFICIENT-MEASUREMENTS', + learning_pass_claimed=False, engineering_scope=True,normal_budget_authorized=False,execution_authorized=False) + self.integrity = dict(complete=True,AP={'pass_all':True},prior={'synthetic':'prior checks'},anchors={'synthetic':'b'*64}, + old_experiments_reexecuted=False,old_files_modified=False,five_frozen_reference_trials_rerun_here=False, + scientific_learning_pass_claimed=False) + self.final = dict(schema='conditioning-execution-final-v1',status=status,source_code_commit='a'*40, + numerical_and_budget_recomputed_at_seal=True,actual_new_brain_calls=23,actual_new_brain_ticks=125200, + actual_execution_child_processes=31,historical_integrity_after_pass=True, + historical_five_trial_regression='AM sealed evidence preserved; not rerun here', + scientific_acquisition_executed=False,scientific_acquisition_authorized=False,candidate_generated=False, + learning_pass_claimed=False,entire_model_directory_sealed=False,bulk_data_git_backup_claimed=False) + put(self.exp / 'runs/one/state.npz', b'binary-not-exported') + put(self.exp / 'extension-build/info.json', {'not':'small evidence despite suffix'}) + put(self.exp / 'protocol/full-trial-plan.json', {'not':'exported despite small size'}) + put(self.exp / 'logs/one.stdout.log', b'actual fixture log') + self.refresh() + + def refresh(self): + regsha = put(self.exp / P.REGISTRATION,self.reg) + for value in (self.final,self.numerical,self.budget,self.execution): value['registration_sha256']=regsha + self.numerical['execution_complete_sha256']=put(self.exp/'audit/execution-complete.json',self.execution) + self.final['numerical_sha256']=put(self.exp/P.REPORTS['numerical_sha256'],self.numerical) + self.final['budget_sha256']=put(self.exp/P.REPORTS['budget_sha256'],self.budget) + self.final['integrity_after_sha256']=put(self.exp/P.REPORTS['integrity_after_sha256'],self.integrity) + self.final['integrity_at_seal_sha256']=put(self.exp/P.REPORTS['integrity_at_seal_sha256'],self.integrity) + finalsha=put(self.exp/P.FINAL,self.final);seal(self.exp) + self.anchors=(regsha,P.digest(self.exp/'manifest.json'),finalsha) + + def publish(self, anchors=None): + return P.publish(*(self.anchors if anchors is None else anchors),root=self.root) + + def test_complete_small_registry_dynamic_sources_omissions_and_seal(self): + self.fixture(source_count=5); result=self.publish() + self.assertEqual(result['source_count'],5); self.assertTrue(result['publication_complete']) + receipt=read(self.registry/'publication-receipt.json') + self.assertFalse(receipt['complete_bulk_backup']);self.assertFalse(receipt['learning_achieved']) + self.assertEqual(receipt['declared_budget_status'],'INSUFFICIENT-MEASUREMENTS') + self.assertFalse(receipt['normal_training_or_budget_authorized']) + self.assertTrue(P.load_helper(self.root/P.HELPER_REL).verify_seal(self.registry)['pass_all']) + inv=read(self.registry/'artifact-inventory.json')['files'] + omitted={r['path']:r for r in inv if r['disposition']=='omitted'} + self.assertTrue({'runs/one/state.npz','extension-build/info.json','protocol/full-trial-plan.json'} <= set(omitted)) + self.assertTrue(all('sha256' in r and 'bytes' in r for r in omitted.values())) + self.assertFalse((self.registry/'evidence/runs/one/state.npz').exists()) + with self.assertRaisesRegex(ValueError,'never overwrite'): self.publish() + + def test_FAIL_and_INCOMPLETE_preserved_without_promotion(self): + for status in ('FAIL-NUMERICAL-OR-LINEAGE','INCOMPLETE-CHANGED-EDGE-COVERAGE'): + with self.subTest(status=status),tempfile.TemporaryDirectory() as name: + saved=(self.root,self.exp,self.model,self.registry) + self.root=Path(name).resolve();self.exp,self.model,self.registry=(self.root/x for x in (P.EXP_REL,P.MODEL_REL,P.REGISTRY_REL)) + put(self.root/P.HELPER_REL,HELPER.read_bytes());self.fixture(status) + result=self.publish();self.assertEqual(result['declared_audit_status'],status) + self.assertFalse(result['scientific_gate_certified_by_publication']) + self.root,self.exp,self.model,self.registry=saved + + def test_three_external_anchors_and_helper_pin_are_required(self): + self.fixture() + for index in range(3): + bad=list(self.anchors);bad[index]='0'*64 + with self.subTest(index=index),self.assertRaisesRegex(ValueError,'External anchor'): self.publish(bad) + put(self.root/P.HELPER_REL,b'not original') + with self.assertRaisesRegex(ValueError,'helper SHA'):self.publish() + + def test_missing_final_seal_controls_and_partial_workflow_rejected(self): + self.fixture();(self.exp/'checksums.sha256').unlink() + with self.assertRaisesRegex(ValueError,'Missing/nonregular'):self.publish() + self.execution['complete']=False;self.refresh() + with self.assertRaisesRegex(ValueError,'complete recorded execution'):self.publish() + self.assertFalse(self.registry.exists()) + + def test_current_source_and_snapshot_mutations_rejected(self): + self.fixture();source=self.model/'new_source_0.py';old=source.read_bytes();source.write_bytes(b'changed') + with self.assertRaisesRegex(ValueError,'bytes differ'):self.publish() + source.write_bytes(old);put(self.exp/P.SNAPSHOT/'new_source_0.py',b'changed snapshot');seal(self.exp) + anchors=(self.anchors[0],P.digest(self.exp/'manifest.json'),self.anchors[2]) + with self.assertRaisesRegex(ValueError,'bytes differ'):self.publish(anchors) + + def test_source_escape_or_extra_snapshot_rejected_even_resealed(self): + self.fixture();key=str(self.model/'new_source_0.py') + self.reg['source_sha256'][str(self.root/'elsewhere/new_source_0.py')]=self.reg['source_sha256'].pop(key) + self.refresh() + with self.assertRaisesRegex(ValueError,'escapes exact AQ'):self.publish() + self.fixture();put(self.exp/P.SNAPSHOT/'unregistered.py',b'extra');seal(self.exp) + with self.assertRaisesRegex(ValueError,'snapshot inventory'): + self.publish((self.anchors[0],P.digest(self.exp/'manifest.json'),self.anchors[2])) + + def test_all_linked_report_SHAs_checked(self): + self.fixture() + for field in P.REPORTS: + original=self.final[field];self.final[field]='0'*64;put(self.exp/P.FINAL,self.final);seal(self.exp) + with self.subTest(field=field),self.assertRaisesRegex(ValueError,'linked report SHA'): + self.publish((self.anchors[0],P.digest(self.exp/'manifest.json'),P.digest(self.exp/P.FINAL))) + self.final[field]=original + + def test_historical_prior_disagreement_and_learning_overclaim_rejected(self): + self.fixture();changed=copy.deepcopy(self.integrity);changed['prior']['synthetic']='different' + self.final['integrity_at_seal_sha256']=put(self.exp/P.REPORTS['integrity_at_seal_sha256'],changed) + put(self.exp/P.FINAL,self.final);seal(self.exp) + with self.assertRaisesRegex(ValueError,'integrity reports disagree'): + self.publish((self.anchors[0],P.digest(self.exp/'manifest.json'),P.digest(self.exp/P.FINAL))) + self.final['learning_pass_claimed']=True;self.refresh() + with self.assertRaisesRegex(ValueError,'scientific/backup claim'):self.publish() + + def test_process_reference_rebinding_rejected_even_resealed(self): + self.fixture();self.numerical['process_evidence']['process-0']['worker_sha256']='0'*64;self.refresh() + with self.assertRaisesRegex(ValueError,'process file reference'):self.publish() + + def test_oversize_required_evidence_and_unlisted_payload_rejected(self): + self.fixture();put(self.exp/'extra.json',{} ) + with self.assertRaisesRegex(ValueError,'unlisted'):self.publish() + (self.exp/'extra.json').unlink() + with patch.object(P,'MAX_BYTES',100): + with self.assertRaisesRegex(ValueError,'oversize'):self.publish() + + def test_postcopy_omitted_bulk_mutation_preserves_partial_registry(self): + self.fixture();helper=P.load_helper(self.root/P.HELPER_REL);original=helper._copy_verified;changed=False + def copying(*args): + nonlocal changed + original(*args) + if not changed: + changed=True;put(self.exp/'runs/one/state.npz',b'mutated omitted file') + with patch.object(P,'load_helper',return_value=helper),patch.object(helper,'_copy_verified',side_effect=copying): + with self.assertRaisesRegex(ValueError,'Payload byte count|Payload SHA'):self.publish() + self.assertTrue(read(self.registry/'publication-failure.json')['partial_outputs_preserved']) + self.assertFalse((self.registry/'publication-receipt.json').exists()) + with self.assertRaisesRegex(ValueError,'never overwrite'):self.publish() + + +if __name__=='__main__':unittest.main()