Digital Drosophila Archive
288 lines
18 KiB
Python
288 lines
18 KiB
Python
"""Exclusive small-evidence publication of an already sealed AP experiment.
|
|
|
|
Three external hashes are required. Publication verifies bytes and reporting
|
|
identities, not the numerical gate or learning. The imported AO trajectory and
|
|
other old bulk remain external; this tool does not reclassify AO's failure.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import sys
|
|
|
|
|
|
ROOT = Path('/data/digital-drosophila')
|
|
MODEL_REL = Path('models/male-cns-lif/conditioning-interface-v2')
|
|
EXP_REL = Path('experiments/phase4ap-conditioning-interface-v2')
|
|
REGISTRY_REL = Path('registry/phase4ap-conditioning-interface-v2')
|
|
AO_REL = Path('experiments/phase4ao-conditioning-interface')
|
|
HELPER_REL = Path('models/male-cns-lif/pn-recruitment-audit-v1/publication.py')
|
|
HELPER_SHA = '7a79c499098b74163e7bb16adabcbc54e7069dafa204d0b99c7f4c03d79f8016'
|
|
SOURCE_FILES = frozenset(('coordinator.py', 'test_coordinator.py', 'isolated_workers.py',
|
|
'test_isolated_workers.py', 'PROTOCOL.md'))
|
|
REGISTRATION, FINAL = 'protocol/preregistered.json', 'audit/final-status.json'
|
|
ANALYSES = ('audit/interface-analysis.json', 'audit/failure-analysis.json')
|
|
SNAPSHOT = 'protocol/source-snapshot/'
|
|
FIRST = 'D1--reference--epoch-000'
|
|
AO_STATUS = 'FAIL-POSTPROCESS-MODULE-IDENTITY'
|
|
MAX_BYTES = 4 * 1024 ** 2
|
|
SUFFIXES = frozenset(('.json', '.md', '.py', '.log', '.sha256'))
|
|
SHA = re.compile(r'^[0-9a-f]{64}$')
|
|
|
|
|
|
def require(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
|
|
def digest(path):
|
|
value = hashlib.sha256()
|
|
with Path(path).open('rb') as stream:
|
|
for chunk in iter(lambda: stream.read(8 << 20), b''):
|
|
value.update(chunk)
|
|
return value.hexdigest()
|
|
|
|
|
|
def load_helper(path):
|
|
raw = Path(path).read_bytes()
|
|
require(hashlib.sha256(raw).hexdigest() == HELPER_SHA, 'Frozen helper SHA mismatch')
|
|
spec = importlib.util.spec_from_file_location('phase4ap_pinned_seal_helper', path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
exec(compile(raw, str(path), 'exec'), module.__dict__)
|
|
return module
|
|
|
|
|
|
def write_json(path, value):
|
|
raw = (json.dumps(value, indent=2, sort_keys=True, 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, 'Missing/oversize required evidence: ' + name)
|
|
path = helper._regular_file(experiment, name)
|
|
require(digest(path) == rows[name]['sha256'], 'Evidence changed: ' + name)
|
|
value = helper._json_bytes(path.read_bytes())
|
|
require(type(value) is dict, 'Expected JSON object: ' + name)
|
|
return value
|
|
|
|
|
|
def source_identity(helper, root, experiment, rows, registration, reg_sha):
|
|
model = root / MODEL_REL
|
|
sources, prepared = registration.get('source_sha256'), registration.get('prepared_file_sha256')
|
|
require(type(sources) is dict and set(sources) == {str(model / name) for name in SOURCE_FILES} and
|
|
type(prepared) is dict, 'Registered AP source inventory must be exactly five fixed files')
|
|
proof = []
|
|
for name in sorted(SOURCE_FILES):
|
|
checksum = sources[str(model / name)]
|
|
require(type(checksum) is str and SHA.fullmatch(checksum), 'Malformed source SHA')
|
|
relative = SNAPSHOT + name
|
|
original = helper._regular_file(root, (MODEL_REL / name).as_posix())
|
|
snapshot = helper._regular_file(experiment, relative)
|
|
require(relative in rows and rows[relative]['sha256'] == checksum and
|
|
prepared.get(str(snapshot)) == checksum and rows[relative]['bytes'] <= MAX_BYTES and
|
|
original.stat().st_size == rows[relative]['bytes'] and digest(original) == digest(snapshot) == checksum,
|
|
'Registered source, prepared snapshot and current model differ: ' + name)
|
|
proof.append(dict(model_source=(MODEL_REL / name).as_posix(), sealed_source=relative,
|
|
copy='source-snapshot/' + name, bytes=rows[relative]['bytes'], sha256=checksum))
|
|
require({name for name in rows if name.startswith(SNAPSHOT)} == {SNAPSHOT + name for name in SOURCE_FILES},
|
|
'Sealed source snapshot inventory differs from registered five')
|
|
for text, checksum in prepared.items():
|
|
require(type(text) is str and type(checksum) is str and SHA.fullmatch(checksum), 'Invalid prepared identity')
|
|
path = Path(text)
|
|
require(path.is_absolute() and path.is_relative_to(experiment) and '..' not in path.parts,
|
|
'Prepared path escapes AP')
|
|
relative = path.relative_to(experiment).as_posix()
|
|
helper._relative_name(relative)
|
|
require(relative in rows and rows[relative]['sha256'] == checksum, 'Prepared file differs from seal: ' + relative)
|
|
commit = registration.get('source_code_commit')
|
|
require(type(commit) is str and re.fullmatch('[0-9a-f]{40}', commit), 'Missing registered source commit')
|
|
return dict(registration_sha256=reg_sha, source_count=5, source_code_commit=commit, files=proof,
|
|
model_snapshot_registration_exact=True, git_blobs_reverified_by_publisher=False,
|
|
entire_model_directory_sealed=False)
|
|
|
|
|
|
def declared_scope(root, reg, final, analysis, reg_sha):
|
|
require(reg.get('schema') == 'phase4ap-process-isolated-interface-v1', 'Wrong registration phase')
|
|
expected = [f'{arm}--{mode}--epoch-{epoch:03d}' for arm in ('D1', 'D0', 'frozen')
|
|
for mode in ('reference', 'light') for epoch in (0, 1)]
|
|
slots = reg.get('slots')
|
|
require(type(slots) is list and [row.get('row_id') for row in slots] == expected, 'Wrong registered slot coverage/order')
|
|
for row in slots:
|
|
name = row['row_id']
|
|
imported = name == FIRST
|
|
base = root / (AO_REL if imported else EXP_REL)
|
|
require(row.get('origin') == ('imported_sealed_AO_neural_trajectory' if imported else 'new_AP_neural_execution') and
|
|
row.get('neural_dir') == str(base / 'runs' / name) and row.get('case_dir') == str(base / 'cases' / name) and
|
|
row.get('original_case_dir') == str(root / AO_REL / 'cases' / name) and
|
|
type(row.get('steps')) is int and row['steps'] == 1200, 'Wrong imported/new slot provenance')
|
|
for key, number in dict(covered_slots=12, new_neural_calls=11, covered_ticks=14400,
|
|
new_neural_ticks=13200, new_sampler_calls=0).items():
|
|
require(type(reg.get(key)) is int and reg[key] == number, 'Wrong registered budget/coverage: ' + key)
|
|
require(reg.get('imported_slot') == FIRST and reg.get('original_failed_workflow_reclassified') is False and
|
|
reg.get('original_AO_process_complete') is False and reg.get('inherited_neural_checkpoint_complete') is True,
|
|
'Registration reclassifies or obscures imported AO origin')
|
|
pin = reg.get('imported_checkpoint_sha256')
|
|
require(type(pin) is str and SHA.fullmatch(pin) and type(reg.get('anchors')) is dict and
|
|
reg['anchors'].get(str(root / AO_REL / 'runs' / FIRST / 'checkpoint-manifest.json')) == pin,
|
|
'Imported checkpoint is not bound to its registered anchor')
|
|
require(reg.get('full_conditioning') is False and reg.get('learning_success_claimed') is False and
|
|
reg.get('within_trial_updates') is False and reg.get('retention_transition_supported') is False,
|
|
'Registered interface scope changed')
|
|
status = final.get('status')
|
|
require(type(status) is str and re.fullmatch('(PASS|FAIL|INCOMPLETE)(-[A-Z0-9]+)*', status), 'Unknown final status')
|
|
require(final.get('registration_sha256') == analysis.get('registration_sha256') == reg_sha and
|
|
analysis.get('status') == status, 'Final/analysis/registration identity mismatch')
|
|
for value in (final, analysis):
|
|
for flag in ('learned_candidate', 'learning_success_claimed', 'full_conditioning_executed'):
|
|
require(value.get(flag) is False, 'No-learning scope must be explicit: ' + flag)
|
|
require(final.get('AO_failure_preserved') is True and analysis.get('original_AO_status') == AO_STATUS,
|
|
'AP must not reclassify AO as successful')
|
|
# Report consistency only. Do not rerun the numerical gate or upgrade FAIL.
|
|
declared = {}
|
|
limits = dict(covered_slots=12, new_neural_calls=11, imported_neural_trajectories=1, new_sampler_calls=0)
|
|
for key, limit in limits.items():
|
|
a, b = final.get(key), analysis.get(key)
|
|
require(type(a) is int and type(b) is int and a == b and 0 <= a <= limit, 'Inconsistent declared count: ' + key)
|
|
require(status.startswith('FAIL') or a == limit, 'Nonfailed finalized interface lacks declared coverage')
|
|
declared[key] = a
|
|
return dict(original_AO_status=AO_STATUS, original_AO_process_complete=False,
|
|
planned_covered_slots=12, planned_imported_neural_trajectories=1, planned_new_neural_calls=11,
|
|
declared_final_counts=declared, imported_checkpoint_sha256=pin,
|
|
scope='Declared report/registration consistency only; imported AO bulk not reread by publisher',
|
|
scientific_gate_recomputed=False, imported_bulk_independently_reverified=False)
|
|
|
|
|
|
def selection(row):
|
|
name = row['path']; parts = PurePosixPath(name)
|
|
if parts.name == 'full-trial-plan.json':
|
|
return None, 'full_trial_plan_not_exported'
|
|
if any(part in ('extension-build', 'build', '__pycache__') for part in parts.parts):
|
|
return None, 'build_or_cache_artifact'
|
|
if row['bytes'] > MAX_BYTES:
|
|
return None, 'exceeds_4_MiB'
|
|
if name.startswith(SNAPSHOT):
|
|
return 'source-snapshot/' + parts.name, None
|
|
if parts.suffix.lower() in SUFFIXES:
|
|
return 'evidence/' + name, None
|
|
return None, 'binary_or_non_selected_extension'
|
|
|
|
|
|
def file_rows(directory):
|
|
return [dict(path=p.relative_to(directory).as_posix(), bytes=p.stat().st_size, sha256=digest(p))
|
|
for p in sorted(directory.rglob('*')) if p.is_file()]
|
|
|
|
|
|
def seal_registry(helper, registry):
|
|
files = file_rows(registry)
|
|
write_json(registry / 'manifest.json', dict(root=str(registry), files=files))
|
|
raw = ''.join(row['sha256'] + ' ' + row['path'] + '\n' for row in files)
|
|
raw += digest(registry / 'manifest.json') + ' manifest.json\n'
|
|
require(len(raw.encode()) <= MAX_BYTES, 'Small registry checksum list exceeds 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):
|
|
"""Verify a finalized AP seal and create only its absent small registry."""
|
|
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 are mandatory')
|
|
root = Path(root)
|
|
require(root.is_absolute() and root.is_dir() and not root.is_symlink(), 'Invalid project root')
|
|
root = root.resolve(strict=True)
|
|
experiment, registry, helper_path = (root / rel for rel in (EXP_REL, REGISTRY_REL, HELPER_REL))
|
|
for path in (experiment, registry, root / MODEL_REL, helper_path):
|
|
cursor = root
|
|
for part in path.relative_to(root).parts:
|
|
cursor /= part
|
|
require(not cursor.is_symlink(), 'Symlink in 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 = {row['path']: row for row 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 = small_json(helper, experiment, rows, REGISTRATION)
|
|
final = small_json(helper, experiment, rows, FINAL)
|
|
linked = [name for name in ANALYSES if name in rows and rows[name]['sha256'] == final.get('analysis_sha256')]
|
|
require(len(linked) == 1, 'Final must name one sealed interface/failure analysis SHA')
|
|
analysis_path = linked[0]
|
|
require('analysis_path' not in final or final['analysis_path'] == analysis_path, 'Analysis path/SHA conflict')
|
|
analysis = small_json(helper, experiment, rows, analysis_path)
|
|
scope = declared_scope(root, reg, final, analysis, registration_sha256)
|
|
sources = source_identity(helper, root, experiment, rows, reg, registration_sha256)
|
|
inventory, plans = [], []
|
|
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: plans.append((row, target))
|
|
targets = [target for _, target in plans]
|
|
required = {'evidence/' + name for name in (REGISTRATION, FINAL, analysis_path, 'manifest.json', 'checksums.sha256')}
|
|
required |= {'source-snapshot/' + name for name in SOURCE_FILES}
|
|
require(len(set(targets)) == len(targets) and required <= set(targets), 'Missing/oversize required evidence or destination collision')
|
|
registry.parent.mkdir(parents=True, exist_ok=True); registry.mkdir(exist_ok=False)
|
|
try:
|
|
records = []
|
|
for row, target in plans:
|
|
helper._copy_verified(helper._regular_file(experiment, row['path']), registry / target, row['bytes'], row['sha256'])
|
|
records.append(dict(source=(EXP_REL / row['path']).as_posix(), copy=target, bytes=row['bytes'], sha256=row['sha256']))
|
|
require(helper.verify_seal(experiment) == verified, 'AP experiment changed while publishing')
|
|
require(source_identity(helper, root, experiment, rows, reg, registration_sha256) == sources, 'Registered AP source changed')
|
|
require(digest(helper_path) == HELPER_SHA and digest(__file__) == own_sha, 'Publisher/helper changed')
|
|
write_json(registry / 'evidence-index.json', records)
|
|
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(experiment_manifest_sha256=experiment_manifest_sha256,
|
|
files=inventory, 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_AP_payload_and_controls=True, complete_bulk_backup=False,
|
|
imported_AO_bulk_is_external=True))
|
|
receipt = dict(schema='phase4ap-publication-v1', publication_complete=True, registry=REGISTRY_REL.as_posix(),
|
|
project_root=str(root), declared_audit_status=final['status'], registration_sha256=registration_sha256,
|
|
experiment_manifest_sha256=experiment_manifest_sha256, experiment_checksums_sha256=verified['checksums_sha256'],
|
|
final_sha256=final_sha256, analysis_path=analysis_path, analysis_sha256=final['analysis_sha256'],
|
|
source_count=5, publisher_sha256=own_sha, helper_sha256=HELPER_SHA, helper_globals_modified=False,
|
|
files=file_rows(registry), copied_evidence_files=len(records), copied_evidence_bytes=sum(r['bytes'] for r in records),
|
|
source_origin=scope, bulk_remains_at=EXP_REL.as_posix(), imported_bulk_remains_at=AO_REL.as_posix(),
|
|
complete_bulk_backup=False, 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,
|
|
receipt_self_hash='Returned externally; also covered by the new registry seal.',
|
|
registry_controls='Created after this receipt, independently verified; no circular self hash.',
|
|
limits=dict(max_selected_file_bytes=MAX_BYTES, selected_suffixes=sorted(SUFFIXES)))
|
|
write_json(registry / 'publication-receipt.json', receipt)
|
|
sealed = seal_registry(helper, registry)
|
|
return dict(publication_complete=True, declared_audit_status=final['status'], registry=str(registry), source_count=5,
|
|
receipt_sha256=digest(registry / 'publication-receipt.json'), registry_manifest_sha256=sealed['manifest_sha256'],
|
|
registry_checksums_sha256=sealed['checksums_sha256'], copied_evidence_files=len(records),
|
|
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
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
for name in ('registration-sha256', 'experiment-manifest-sha256', 'final-sha256'):
|
|
parser.add_argument('--' + name, required=True)
|
|
args = parser.parse_args()
|
|
print(json.dumps(publish(args.registration_sha256, args.experiment_manifest_sha256, args.final_sha256),
|
|
sort_keys=True, ensure_ascii=False, allow_nan=False), flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|