qishizhe/digital-drosophila
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Seal Phase 4AP isolated delta interface PASS, not learning

Browse Source
This commit is contained in:
Digital Drosophila Archive
2026-09-30 18:36:01 +08:00
parent 0bcce98332
commit b33e7ea047
360 changed files with 127513 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/phase4ap-publication/publish_phase4ap.py
+287
View File
@@ -0,0 +1,287 @@
"""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()
tools/phase4ap-publication/test_publish_phase4ap.py
+290
View File
@@ -0,0 +1,290 @@
"""CPU-only temporary evidence trees; no real AP/brain execution."""
import json
from pathlib import Path
import tempfile
import unittest
from unittest import mock
import publish_phase4ap as P
_candidates = (
Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py',
Path(__file__).resolve().parents[2] / P.HELPER_REL,
)
HELPER = next((path for path in _candidates if path.is_file()), _candidates[0])
def put(path, value):
path.parent.mkdir(parents=True, exist_ok=True)
raw = value if isinstance(value, 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(directory):
files = [r for r in P.file_rows(directory) if r['path'] not in ('manifest.json', 'checksums.sha256')]
put(directory / 'manifest.json', dict(root=str(directory), files=files))
raw = ''.join(row['sha256'] + ' ' + row['path'] + '\n' for row in files)
put(directory / 'checksums.sha256', (raw + P.digest(directory / '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.registry, self.model = (self.root / rel for rel in (P.EXP_REL, P.REGISTRY_REL, P.MODEL_REL))
put(self.root / P.HELPER_REL, HELPER.read_bytes())
def fixture(self, status='PASS-EXACT-PROCESS-ISOLATED-DELTA-INTERFACE-NOT-LEARNING'):
sources, prepared = {}, {}
for name in P.SOURCE_FILES:
raw = ('# synthetic only: ' + name).encode()
checksum = put(self.model / name, raw)
snapshot = self.exp / P.SNAPSHOT / name
put(snapshot, raw)
sources[str(self.model / name)] = prepared[str(snapshot)] = checksum
slots = []
for arm in ('D1', 'D0', 'frozen'):
for mode in ('reference', 'light'):
for epoch in (0, 1):
name = f'{arm}--{mode}--epoch-{epoch:03d}'
imported = name == P.FIRST
base = self.root / (P.AO_REL if imported else P.EXP_REL)
slots.append(dict(row_id=name, steps=1200, origin='imported_sealed_AO_neural_trajectory' if imported else 'new_AP_neural_execution',
neural_dir=str(base / 'runs' / name), case_dir=str(base / 'cases' / name),
original_case_dir=str(self.root / P.AO_REL / 'cases' / name)))
self.reg = dict(schema='phase4ap-process-isolated-interface-v1', source_sha256=sources,
prepared_file_sha256=prepared, source_code_commit='5' * 40, slots=slots,
covered_slots=12, new_neural_calls=11, covered_ticks=14400, new_neural_ticks=13200, new_sampler_calls=0,
imported_slot=P.FIRST, imported_checkpoint_sha256='e' * 64,
anchors={str(self.root / P.AO_REL / 'runs' / P.FIRST / 'checkpoint-manifest.json'): 'e' * 64},
original_failed_workflow_reclassified=False, original_AO_process_complete=False,
inherited_neural_checkpoint_complete=True, full_conditioning=False, learning_success_claimed=False,
within_trial_updates=False, retention_transition_supported=False)
self.analysis = dict(status=status, covered_slots=12, new_neural_calls=11,
imported_neural_trajectories=1, new_sampler_calls=0, original_AO_status=P.AO_STATUS,
learned_candidate=False, learning_success_claimed=False, full_conditioning_executed=False)
self.analysis_path = P.ANALYSES[1 if status.startswith('FAIL') else 0]
self.final = {k: v for k, v in self.analysis.items() if k != 'original_AO_status'}
self.final['AO_failure_preserved'] = True
put(self.exp / 'logs/first.log', b'complete evidence, not learning\n')
put(self.exp / 'runs/new/state.npz', b'bulk remains here')
put(self.exp / 'runs/new/hash.jsonl', b'{}\n')
put(self.exp / 'extension-build/source.py', b'build source not evidence')
put(self.exp / 'large.json', b' ' * (P.MAX_BYTES + 1))
put(self.exp / 'protocol/full-trial-plan.json', {'not_exported_even_if_small': True})
self.refresh()
def refresh(self):
reg_sha = put(self.exp / P.REGISTRATION, self.reg)
self.analysis['registration_sha256'] = reg_sha
self.final['registration_sha256'] = reg_sha
self.final['analysis_sha256'] = put(self.exp / self.analysis_path, self.analysis)
put(self.exp / P.FINAL, self.final)
seal(self.exp)
self.anchors = (reg_sha, P.digest(self.exp / 'manifest.json'), P.digest(self.exp / P.FINAL))
def publish(self, anchors=None):
return P.publish(*(anchors or self.anchors), root=self.root)
def test_success_independent_seal_exact_copy_and_omitted_inventory(self):
self.fixture()
prior = P.file_rows(self.exp)
result = self.publish()
self.assertTrue(result['publication_complete'])
self.assertEqual(result['source_count'], 5)
self.assertFalse(result['complete_bulk_backup'])
helper = P.load_helper(self.root / P.HELPER_REL)
registry_seal = helper.verify_seal(self.registry)
self.assertEqual(registry_seal['manifest_sha256'], result['registry_manifest_sha256'])
self.assertEqual(result['receipt_sha256'], P.digest(self.registry / 'publication-receipt.json'))
receipt = read(self.registry / 'publication-receipt.json')
self.assertEqual(receipt['source_origin']['declared_final_counts']['new_neural_calls'], 11)
self.assertEqual(receipt['source_origin']['original_AO_status'], P.AO_STATUS)
self.assertFalse(receipt['source_origin']['imported_bulk_independently_reverified'])
self.assertFalse(receipt['learning_achieved'])
for row in receipt['files']:
self.assertEqual(P.digest(self.registry / row['path']), row['sha256'])
inventory = read(self.registry / 'artifact-inventory.json')['files']
self.assertEqual({r['path'] for r in prior}, {r['path'] for r in inventory})
for row in inventory:
self.assertEqual(row['sha256'], P.digest(self.exp / row['path']))
indexed = {r['path']: r for r in inventory}
for name in ('runs/new/state.npz', 'runs/new/hash.jsonl', 'extension-build/source.py',
'large.json', 'protocol/full-trial-plan.json'):
self.assertEqual(indexed[name]['disposition'], 'omitted')
self.assertTrue((self.registry / 'source-snapshot/coordinator.py').is_file())
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(prior, P.file_rows(self.exp))
def test_incomplete_and_failed_status_are_not_upgraded(self):
for status in ('INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE', 'FAIL-CHILD-PROCESS'):
with self.subTest(status=status):
# Separate exclusive trees; no deletion/retry of a publication.
with tempfile.TemporaryDirectory() as temp:
self.root = Path(temp).resolve()
self.exp, self.registry, self.model = (self.root / rel for rel in (P.EXP_REL, P.REGISTRY_REL, P.MODEL_REL))
put(self.root / P.HELPER_REL, HELPER.read_bytes())
self.fixture(status)
if status.startswith('FAIL'):
for value in (self.final, self.analysis):
value.update(covered_slots=1, new_neural_calls=0)
self.refresh()
result = self.publish()
self.assertEqual(result['declared_audit_status'], status)
self.assertFalse(result['scientific_gate_certified_by_publication'])
def test_three_anchors_are_required_and_rechecked(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.assertRaisesRegex(ValueError, 'mandatory'):
self.publish(['', self.anchors[1], self.anchors[2]])
self.assertFalse(self.registry.exists())
def test_missing_final_or_unsealed_tree_refuses_publication(self):
self.fixture()
(self.exp / P.FINAL).unlink()
with self.assertRaises(ValueError):
self.publish()
put(self.exp / P.FINAL, self.final)
(self.exp / 'checksums.sha256').unlink()
with self.assertRaises(ValueError):
self.publish()
self.assertFalse(self.registry.exists())
def test_existing_registry_is_untouched(self):
self.fixture()
marker = self.registry / 'keep.txt'; put(marker, b'keep')
with self.assertRaisesRegex(ValueError, 'never overwrite'):
self.publish()
self.assertEqual(marker.read_bytes(), b'keep')
self.assertEqual(len(list(self.registry.iterdir())), 1)
def test_source_mutation_and_extra_or_missing_inventory_fail(self):
self.fixture()
put(self.model / 'coordinator.py', b'# different')
with self.assertRaisesRegex(ValueError, 'differ'):
self.publish()
put(self.model / 'coordinator.py', (self.exp / P.SNAPSHOT / 'coordinator.py').read_bytes())
put(self.exp / P.SNAPSHOT / 'extra.py', b'# extra')
seal(self.exp)
anchors = (self.anchors[0], P.digest(self.exp / 'manifest.json'), self.anchors[2])
with self.assertRaisesRegex(ValueError, 'inventory differs'):
self.publish(anchors)
def test_resealed_snapshot_disagrees_with_registered_source(self):
self.fixture()
put(self.exp / P.SNAPSHOT / 'coordinator.py', b'# altered snapshot')
seal(self.exp)
with self.assertRaisesRegex(ValueError, 'differ'):
self.publish((self.anchors[0], P.digest(self.exp / 'manifest.json'), self.anchors[2]))
def test_import_cannot_be_reclassified_as_new_or_AO_success(self):
self.fixture()
self.reg['slots'][0]['origin'] = 'new_AP_neural_execution'; self.refresh()
with self.assertRaisesRegex(ValueError, 'slot provenance'):
self.publish()
self.reg['slots'][0]['origin'] = 'imported_sealed_AO_neural_trajectory'
self.analysis['original_AO_status'] = 'PASS'; self.refresh()
with self.assertRaisesRegex(ValueError, 'reclassify AO'):
self.publish()
def test_missing_duplicate_slot_and_source_escape_fail(self):
self.fixture()
self.reg['slots'][-1] = dict(self.reg['slots'][0]); self.refresh()
with self.assertRaisesRegex(ValueError, 'coverage/order'):
self.publish()
self.reg['slots'][-1]['row_id'] = 'frozen--light--epoch-001'
self.reg['slots'][-1]['origin'] = 'new_AP_neural_execution'
self.reg['slots'][-1]['neural_dir'] = str(self.exp / 'runs/frozen--light--epoch-001')
self.reg['slots'][-1]['case_dir'] = str(self.exp / 'cases/frozen--light--epoch-001')
self.reg['slots'][-1]['original_case_dir'] = str(self.root / P.AO_REL / 'cases/frozen--light--epoch-001')
key = str(self.model / 'coordinator.py')
self.reg['source_sha256'][str(self.root / 'elsewhere/coordinator.py')] = self.reg['source_sha256'].pop(key)
self.refresh()
with self.assertRaisesRegex(ValueError, 'exactly five'):
self.publish()
def test_learning_claim_or_inconsistent_declared_count_rejected(self):
self.fixture()
self.analysis['learning_success_claimed'] = True; self.refresh()
with self.assertRaisesRegex(ValueError, 'No-learning'):
self.publish()
self.analysis['learning_success_claimed'] = False
self.analysis['new_neural_calls'] = 12; self.refresh()
with self.assertRaisesRegex(ValueError, 'declared count'):
self.publish()
def test_final_analysis_link_and_helper_pin_rejected(self):
self.fixture()
self.final['analysis_sha256'] = '0' * 64
put(self.exp / P.FINAL, self.final); seal(self.exp)
anchors = (self.anchors[0], P.digest(self.exp / 'manifest.json'), P.digest(self.exp / P.FINAL))
with self.assertRaisesRegex(ValueError, 'analysis SHA'):
self.publish(anchors)
put(self.root / P.HELPER_REL, b'# not the pinned helper')
with self.assertRaisesRegex(ValueError, 'helper SHA mismatch'):
self.publish(anchors)
def test_unlisted_payload_or_corrupt_checksums_refused(self):
self.fixture()
put(self.exp / 'unlisted.txt', b'not registered')
with self.assertRaisesRegex(ValueError, 'unlisted'):
self.publish()
(self.exp / 'unlisted.txt').unlink()
with (self.exp / 'checksums.sha256').open('ab') as stream:
stream.write(b'bad\n')
with self.assertRaisesRegex(ValueError, 'Malformed checksums'):
self.publish()
def test_postcopy_bulk_change_is_caught_without_deleting_partial_registry(self):
self.fixture()
helper = P.load_helper(self.root / P.HELPER_REL)
original = helper._copy_verified
changed = False
def copy_and_mutate(*args):
nonlocal changed
original(*args)
if not changed:
changed = True
put(self.exp / 'runs/new/state.npz', b'mutated omitted bulk')
with mock.patch.object(helper, '_copy_verified', side_effect=copy_and_mutate), mock.patch.object(P, 'load_helper', return_value=helper):
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()
def test_external_manifest_race_and_registry_mutation_detected(self):
self.fixture()
helper = P.load_helper(self.root / P.HELPER_REL)
original = helper.verify_seal
def new_seal(directory):
put(self.exp / 'logs/first.log', b'changed before full verification')
seal(self.exp)
return original(directory)
with mock.patch.object(helper, 'verify_seal', side_effect=new_seal), mock.patch.object(P, 'load_helper', return_value=helper):
with self.assertRaisesRegex(ValueError, 'differs from external'):
self.publish()
self.assertFalse(self.registry.exists())
self.anchors = (self.anchors[0], P.digest(self.exp / 'manifest.json'), self.anchors[2])
self.publish()
put(self.registry / 'source-snapshot/coordinator.py', b'changed')
with self.assertRaisesRegex(ValueError, 'Payload byte count|Payload SHA'):
helper.verify_seal(self.registry)
if __name__ == '__main__':
unittest.main()