Digital Drosophila Archive
519 lines
30 KiB
Python
519 lines
30 KiB
Python
"""Export CPU reference fixtures from the fixed, committed Phase 4AR prefix.
|
|
|
|
Run with the ORIGINAL models/male-cns-lif/.venv/bin/python -B and --output an
|
|
absolute, nonexistent directory outside frozen input/source directories. This
|
|
tool never launches a brain, resumes training, installs weights or changes AR.
|
|
The original AQ CPU validator is run in fresh CPU-only subprocesses; its only
|
|
new receipts are placed inside the requested output. A second fresh process per
|
|
fixture computes the original U Torch CPU reference, including a warm repeat.
|
|
|
|
Selection is first and last task_id in each available arm among committed
|
|
brain/update/verify triples, determined before result/metric files are opened.
|
|
This bounded engineering sample does not cover all scientific conditions.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
|
|
ROOT = Path('/data/digital-drosophila')
|
|
MODEL = ROOT / 'models/male-cns-lif'
|
|
EXP = ROOT / 'experiments/phase4ar-normal-conditioning'
|
|
ORIGINAL_ENV = MODEL / '.venv'
|
|
REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740'
|
|
SOURCE_COMMIT = '80aeb2093db655329774eff7b32fb6e417b8b40b'
|
|
U_SHA = '96efff678cff910654eb08e86f1a17789e4e5589e340345f419c7027a7d547e2'
|
|
BRIDGE_SHA = '174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61'
|
|
J_SHA = 'abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e'
|
|
ARMS = ('cs_only', 'd_zero', 'dan_only', 'frozen', 'paired', 'random_reward', 'unpaired')
|
|
INPUT_ARRAYS = ('step', 'neuron_idx', 'kc', 'candidate_edge_id', 'candidate_sources',
|
|
'candidate_targets', 'candidate_compartments', 'delta_before', 'base_candidate')
|
|
OUTPUT_ARRAYS = ('delta_after', 'sampled_trace', 'final_trace')
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def canonical(value):
|
|
return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False)
|
|
|
|
|
|
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 unlinked(path):
|
|
path = Path(path)
|
|
require(path.is_absolute() and '..' not in path.parts, 'Absolute nonescaping path required')
|
|
require(not any(p.is_symlink() or (hasattr(p, 'is_junction') and p.is_junction())
|
|
for p in (path, *path.parents)), 'Linked path forbidden: ' + str(path))
|
|
return path
|
|
|
|
|
|
def read(path):
|
|
def pairs(items):
|
|
result = {}
|
|
for key, value in items:
|
|
require(key not in result, 'Duplicate JSON key')
|
|
result[key] = value
|
|
return result
|
|
return json.loads(unlinked(path).read_text(encoding='utf-8'), object_pairs_hook=pairs,
|
|
parse_constant=lambda _: (_ for _ in ()).throw(ValueError('Nonfinite JSON')))
|
|
|
|
|
|
def save(path, value):
|
|
with unlinked(path).open('x', encoding='utf-8', newline='\n') as stream:
|
|
json.dump(value, stream, sort_keys=True, indent=2, ensure_ascii=False, allow_nan=False)
|
|
stream.write('\n')
|
|
|
|
|
|
def identity(path):
|
|
path = unlinked(path)
|
|
require(path.is_file(), 'Missing regular file: ' + str(path))
|
|
size = path.stat().st_size
|
|
result = dict(sha256=digest(path), bytes=size)
|
|
require(path.stat().st_size == size, 'File size changed while hashing')
|
|
return result
|
|
|
|
|
|
def check(path, expected):
|
|
observed = identity(path)
|
|
sha = expected if isinstance(expected, str) else expected['sha256']
|
|
require(observed['sha256'] == sha, 'Pinned bytes changed: ' + str(path))
|
|
if isinstance(expected, dict) and 'bytes' in expected:
|
|
require(observed['bytes'] == expected['bytes'], 'Pinned size changed: ' + str(path))
|
|
return observed
|
|
|
|
|
|
def task_key(task):
|
|
return 'task-' + hashlib.sha256(task['task_id'].encode()).hexdigest()
|
|
|
|
|
|
def select_tasks(tasks, committed_ids):
|
|
"""Only plan metadata and committed entry IDs participate in selection."""
|
|
available = {arm: [] for arm in ARMS}
|
|
seen = set()
|
|
for task in tasks:
|
|
require(task['task_id'] not in seen, 'Duplicate plan task identity')
|
|
seen.add(task['task_id'])
|
|
if task['kind'] != 'train':
|
|
continue
|
|
require(task['arm'] in available, 'Unexpected training arm')
|
|
key = task_key(task)
|
|
if all(key + '--' + stage in committed_ids for stage in ('brain', 'update', 'verify')):
|
|
available[task['arm']].append(task)
|
|
selected = []
|
|
for arm in ARMS:
|
|
ordered = sorted(available[arm], key=lambda task: task['task_id'])
|
|
if ordered:
|
|
selected.append(ordered[0])
|
|
if ordered[-1]['task_id'] != ordered[0]['task_id']:
|
|
selected.append(ordered[-1])
|
|
require(selected, 'No committed train triple is available')
|
|
return selected, {arm: len(available[arm]) for arm in ARMS}
|
|
|
|
|
|
def ledger_snapshot(exp, reg):
|
|
"""Read ledger controls, not all historical outcomes; never instantiate Ledger."""
|
|
control = exp / '.append-ledger'
|
|
require(not (control / 'writer.lock').exists(), 'Ledger writer/stale lock present')
|
|
header_path = control / 'header.json'
|
|
header = read(header_path)
|
|
require(header['schema'] == 'normal-append-ledger-v1' and header['root'] == str(exp)
|
|
and header['registration_sha256'] == REG_SHA, 'Ledger header authority differs')
|
|
expected_initial = {str(Path(p).relative_to(exp)): sha for p, sha in reg['prepared_sha256'].items()}
|
|
expected_initial['protocol/preregistered.json'] = REG_SHA
|
|
require({p: info['sha256'] for p, info in header['initial_inventory'].items()} == expected_initial,
|
|
'Ledger initial inventory differs from registration')
|
|
control_pins = {str(header_path): identity(header_path)}
|
|
tracked = {str(exp / p): info for p, info in header['initial_inventory'].items()}
|
|
entries = {}
|
|
tail = None
|
|
paths = sorted((control / 'entries').iterdir())
|
|
for sequence, path in enumerate(paths, 1):
|
|
require(path.name == f'{sequence:016d}.json', 'Ledger entries must form a contiguous prefix')
|
|
row = read(path)
|
|
name = row['entry_id']
|
|
require(name not in entries and row['schema'] == 'normal-append-ledger-v1-entry'
|
|
and row['registration_sha256'] == REG_SHA and row['sequence'] == sequence
|
|
and row['prev_entry_sha256'] == tail and row['scientific_gate_claimed'] is False,
|
|
'Ledger chain identity differs')
|
|
start_path = control / 'started' / (name + '.json')
|
|
start = read(start_path)
|
|
control_pins[str(start_path)] = check(start_path, row['start_sha256'])
|
|
require(start['schema'] == 'normal-append-ledger-v1-start' and start['entry_id'] == name
|
|
and start['registration_sha256'] == REG_SHA and start['next_sequence'] == sequence
|
|
and start['prev_entry_sha256'] == tail and start['retry_authorized'] is False,
|
|
'Ledger start identity differs')
|
|
for output in row['new_outputs']:
|
|
output_path = unlinked(output['path'])
|
|
require(output_path.is_relative_to(exp) and str(output_path.relative_to(exp)) == output['relative_path']
|
|
and str(output_path) not in tracked, 'Output escapes or repeats immutable history')
|
|
tracked[str(output_path)] = {key: output[key] for key in ('bytes', 'sha256')}
|
|
control_pins[str(path)] = identity(path)
|
|
tail = control_pins[str(path)]['sha256']
|
|
entries[name] = dict(row=row, start=start, ledger_path=str(path), ledger_sha256=tail)
|
|
require(entries, 'Empty committed prefix')
|
|
return entries, tracked, control_pins, dict(entries=len(entries), head_sha256=tail,
|
|
uncommitted_starts=sorted(p.stem for p in (control / 'started').glob('*.json') if p.stem not in entries),
|
|
scope='ledger chain controls and selected artifacts; not a full historical payload audit')
|
|
|
|
|
|
def validate_process(entry, tracked, exp):
|
|
row, start = entry['row'], entry['start']
|
|
name = row['entry_id']
|
|
metadata = row['metadata']
|
|
job_path = unlinked(metadata['job_file'])
|
|
process_path = unlinked(metadata['process_receipt_file'])
|
|
require(job_path == exp / 'jobs' / (name + '.json') and process_path == exp / 'receipts' / name / 'process.json',
|
|
'Process/job location differs')
|
|
require(str(job_path) in tracked and str(process_path) in tracked, 'Process/job not in committed history')
|
|
job_pin = check(job_path, tracked[str(job_path)])
|
|
process_pin = check(process_path, tracked[str(process_path)])
|
|
require(start['job']['path'] == str(job_path) and start['job']['sha256'] == job_pin['sha256'],
|
|
'Started job differs from committed job')
|
|
process = read(process_path)
|
|
require(process['entry_id'] == name and process['registration_sha256'] == REG_SHA
|
|
and process['job_sha256'] == job_pin['sha256'] and process['complete'] is True
|
|
and type(process['exit_code']) is int and process['exit_code'] == 0
|
|
and process['timed_out'] is False and process['exit_observed_by_parent'] is True
|
|
and process.get('launch_error') is None and type(metadata['observed_returncode']) is int
|
|
and metadata['observed_returncode'] == 0, 'Selected process is not committed observed exit zero')
|
|
require(process['argv'][0] == str(ORIGINAL_ENV / 'bin/python') and '-B' in process['argv'],
|
|
'Selected process did not use original -B interpreter')
|
|
job = read(job_path)
|
|
require(job['registration_sha256'] == REG_SHA, 'Selected job registration differs')
|
|
outgoing = {item['path']: item for item in row['new_outputs']}
|
|
require(str(job_path) in outgoing and str(process_path) in outgoing,
|
|
'Job/process must be outputs of this exact committed entry')
|
|
consumed = {item['path']: item['sha256'] for item in row['consumed_inputs']}
|
|
require(all(consumed.get(path) == sha for path, sha in job['input_sha256'].items()),
|
|
'Selected job input pins differ from committed ledger')
|
|
return job, {str(job_path): job_pin, str(process_path): process_pin}, process
|
|
|
|
|
|
def array_evidence(array):
|
|
return dict(dtype=array.dtype.str, shape=list(array.shape), nbytes=int(array.nbytes),
|
|
raw_c_order_sha256=hashlib.sha256(array.tobytes(order='C')).hexdigest())
|
|
|
|
|
|
def load_npz(path):
|
|
import numpy as np
|
|
with np.load(unlinked(path), allow_pickle=False) as archive:
|
|
return {name: archive[name].copy() for name in archive.files}
|
|
|
|
|
|
def save_npz(path, arrays):
|
|
import numpy as np
|
|
with unlinked(path).open('xb') as stream:
|
|
np.savez_compressed(stream, **arrays)
|
|
|
|
|
|
def load_verified_source(name, path, sha):
|
|
raw = unlinked(path).read_bytes()
|
|
require(hashlib.sha256(raw).hexdigest() == sha and name not in sys.modules, 'Unverified source/module binding')
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
exec(compile(raw, str(path), 'exec'), module.__dict__)
|
|
return module
|
|
|
|
|
|
def cpu_environment():
|
|
require(sys.dont_write_bytecode and 'torch' not in sys.modules and 'runtime_light' not in sys.modules,
|
|
'Use a fresh original Python -B process')
|
|
require(Path(sys.prefix) == ORIGINAL_ENV, 'Reference export requires the original registered .venv')
|
|
os.environ['CUDA_VISIBLE_DEVICES'] = ''
|
|
os.environ['PYTHONDONTWRITEBYTECODE'] = '1'
|
|
|
|
|
|
def run_cpu(argv, directory, label):
|
|
started = time.perf_counter()
|
|
environment = dict(os.environ, CUDA_VISIBLE_DEVICES='', PYTHONDONTWRITEBYTECODE='1')
|
|
with (directory / (label + '.stdout.log')).open('xb') as stdout, \
|
|
(directory / (label + '.stderr.log')).open('xb') as stderr:
|
|
process = subprocess.run(argv, cwd=directory, env=environment, stdout=stdout, stderr=stderr,
|
|
timeout=1800, check=False)
|
|
elapsed = time.perf_counter() - started
|
|
save(directory / (label + '-process.json'), dict(argv=argv, exit_code=process.returncode,
|
|
wall_seconds=elapsed, CUDA_VISIBLE_DEVICES='', brain_executed=False))
|
|
require(process.returncode == 0, label + ' failed; partial outputs retained')
|
|
return elapsed
|
|
|
|
|
|
def reference_worker(directory):
|
|
cpu_environment()
|
|
import numpy as np
|
|
directory = unlinked(directory)
|
|
inputs = load_npz(directory / 'inputs.npz')
|
|
command = read(directory / 'command.json')
|
|
sources = read(directory / 'reference-sources.json')
|
|
require(sources['conditioning_update_source']['sha256'] == U_SHA
|
|
and sources['eligibility_bridge_source']['sha256'] == BRIDGE_SHA
|
|
and sources['phase4j_source']['sha256'] == J_SHA, 'Original source pins differ')
|
|
for info in sources.values():
|
|
check(info['path'], info['sha256'])
|
|
update = load_verified_source('_acceleration_original_U', sources['conditioning_update_source']['path'], U_SHA)
|
|
kwargs = {**inputs, **command, 'bridge_source': sources['eligibility_bridge_source']['path'],
|
|
'phase4j_source': sources['phase4j_source']['path']}
|
|
before = {name: array_evidence(value) for name, value in inputs.items()}
|
|
started = time.perf_counter()
|
|
cold = update.update_epoch(**kwargs)
|
|
cold_seconds = time.perf_counter() - started
|
|
started = time.perf_counter()
|
|
warm = update.update_epoch(**kwargs)
|
|
warm_seconds = time.perf_counter() - started
|
|
require(canonical(cold.receipt) == canonical(warm.receipt), 'Cold/warm original receipts differ')
|
|
for name in OUTPUT_ARRAYS:
|
|
require(array_evidence(getattr(cold, name)) == array_evidence(getattr(warm, name)),
|
|
'Cold/warm original output differs: ' + name)
|
|
require(before == {name: array_evidence(value) for name, value in inputs.items()}, 'Reference input mutated')
|
|
import torch
|
|
require(not torch.cuda.is_initialized() and os.environ['CUDA_VISIBLE_DEVICES'] == '', 'CUDA initialized in CPU reference')
|
|
arrays = {'reference_' + name: getattr(cold, name) for name in OUTPUT_ARRAYS}
|
|
commands = [] if command['us_tick'] is None else [command['us_tick']]
|
|
arrays['reference_command_steps'] = np.asarray(commands, dtype=np.int64)
|
|
arrays['reference_snapshots'] = (np.empty((0, len(inputs['kc'])), np.float32) if not commands
|
|
else cold.sampled_trace[None, :].copy())
|
|
save_npz(directory / 'reference.npz', arrays)
|
|
save(directory / 'reference-update-receipt.json', cold.receipt)
|
|
save(directory / 'reference-timing.json', dict(schema='conditioning-original-cpu-timing-v1',
|
|
python_version=sys.version, executable=sys.executable, environment_prefix=sys.prefix,
|
|
numpy_version=np.__version__, torch_version=torch.__version__, cuda_initialized=False,
|
|
cold_update_seconds=cold_seconds, warm_update_seconds=warm_seconds,
|
|
timing_scope='original update_epoch API with loaded arrays, includes source checks/AST parsing; cold includes first Torch import; not pure kernel time',
|
|
process_time_scope='reference-process.json separately includes startup, loading, two updates and output serialization',
|
|
snapshot_scope='exact sole declared-US snapshot equals sampled_trace; empty snapshots for no US',
|
|
output_arrays={name: array_evidence(array) for name, array in arrays.items()},
|
|
source_pins=sources, cold_warm_bitwise_exact=True, brain_executed=False, weights_installed=False))
|
|
|
|
|
|
def validate_destination(output, protected):
|
|
output = unlinked(output)
|
|
require(not output.exists() and output.parent.is_dir(), 'Output must be a fresh directory with an existing parent')
|
|
require(not any(output.is_relative_to(unlinked(parent)) for parent in protected),
|
|
'Output overlaps a frozen input/source directory')
|
|
return output
|
|
|
|
|
|
def export(output):
|
|
cpu_environment()
|
|
started = time.perf_counter()
|
|
own = identity(Path(__file__).absolute())
|
|
regpath = EXP / 'protocol/preregistered.json'
|
|
check(regpath, REG_SHA)
|
|
reg = read(regpath)
|
|
require(reg['schema'] == 'normal-conditioning-registration-v1'
|
|
and reg['source_code_commit'] == SOURCE_COMMIT and reg['scientific_scope'] == 'normal-acquisition-only'
|
|
and reg['subsequent_assays_authorized'] is False, 'Wrong fixed AR authority')
|
|
source_pins = {**reg['source_sha256'], **{p: sha for p, sha in reg['input_sha256'].items()
|
|
if Path(p).is_relative_to(MODEL)}}
|
|
for path, sha in source_pins.items():
|
|
check(path, sha)
|
|
planpath = EXP / 'protocol/normal-plan.json'
|
|
check(planpath, reg['prepared_sha256'][str(planpath)])
|
|
plan = read(planpath)
|
|
entries, tracked, control_pins, snapshot = ledger_snapshot(EXP, reg)
|
|
selected, available = select_tasks(plan['tasks'], set(entries))
|
|
tasks_by_id = {task['task_id']: task for task in plan['tasks']}
|
|
protected = {EXP, MODEL, *[Path(p).parent for p in reg['input_sha256']]}
|
|
output = validate_destination(output, protected)
|
|
output.mkdir()
|
|
selection = dict(schema='conditioning-acceleration-selection-v1', registration_sha256=REG_SHA,
|
|
source_code_commit=SOURCE_COMMIT, rule='first and last lexicographic task_id per arm among committed train brain/update/verify triples',
|
|
metrics_used_for_selection=False, available_train_counts=available, selected_tasks=selected,
|
|
selected_count=len(selected), maximum_cases=14, ledger_snapshot=snapshot,
|
|
coverage_limit='engineering fixtures only; neither complete scientific-condition coverage nor a learning result')
|
|
save(output / 'selection.json', selection)
|
|
save(output / 'original-source-pins.json', source_pins)
|
|
# No task result, update or event array is read before the immutable selection above.
|
|
fixtures = []
|
|
for index, task in enumerate(selected):
|
|
item_started = time.perf_counter()
|
|
fixture_id = f'{index:02d}-{task["arm"]}'
|
|
directory = output / fixture_id
|
|
directory.mkdir()
|
|
key = task_key(task)
|
|
file_pins = {}
|
|
jobs = {}
|
|
process_records = {}
|
|
for stage in ('brain', 'update', 'verify'):
|
|
job, pins, process = validate_process(entries[key + '--' + stage], tracked, EXP)
|
|
jobs[stage] = job
|
|
file_pins.update(pins)
|
|
process_records[stage] = process
|
|
result_path = EXP / 'results' / (key + '.json')
|
|
file_pins[str(result_path)] = check(result_path, tracked[str(result_path)])
|
|
result = read(result_path)
|
|
require(result['schema'] == 'normal-task-result-v1' and result['complete'] is True
|
|
and result['registration_sha256'] == REG_SHA and result['task'] == task, 'Selected result/plan differs')
|
|
authority_path = unlinked(result['epoch_authority']['path'])
|
|
authority_sha = result['epoch_authority']['sha256']
|
|
file_pins[str(authority_path)] = check(authority_path, tracked[str(authority_path)])
|
|
require(file_pins[str(authority_path)]['sha256'] == authority_sha, 'Authority result/ledger pin differs')
|
|
authority = read(authority_path)
|
|
command = dict(authority['command'], n_neurons=166700)
|
|
require(authority['epoch_id'] == task['task_id'] and authority['diagnostic_mode'] == 'light'
|
|
and command == dict(steps=task['steps'], n_neurons=166700, compartment=task['compartment_verbatim'],
|
|
us_tick=task['recipe']['us']['tick'], d_command=task['recipe']['D_command'],
|
|
frozen=not task['recipe']['update_enabled'], trace_backend='torch'), 'Actual command differs from registered task')
|
|
update_dir = unlinked(result['update_dir'])
|
|
require(update_dir == EXP / 'updates' / key and authority_path == EXP / 'jobs' / (key + '.authority.json')
|
|
and authority['checkpoint_dir'] == result['checkpoint_dir'] == str(EXP / 'runs' / key)
|
|
and authority['checkpoint_manifest_sha256'] == result['checkpoint_manifest_sha256'],
|
|
'Task checkpoint/update location differs')
|
|
for stage in ('update', 'verify'):
|
|
require(jobs[stage]['authority_file'] == str(authority_path) and jobs[stage]['authority_sha256'] == authority_sha,
|
|
'Authority not bound to selected successful job')
|
|
require(jobs['brain']['output_dir'] == authority['checkpoint_dir'] and jobs['update']['output_dir'] == str(update_dir)
|
|
and jobs['verify']['update_dir'] == str(update_dir), 'Selected process output differs')
|
|
require(jobs['brain']['kind'] == 'brain' and jobs['update']['kind'] == 'update'
|
|
and jobs['verify']['kind'] == 'cpu_verify', 'Selected process role differs')
|
|
dependency = task['delta_dependency']
|
|
if dependency == dict(kind='zero', source=None):
|
|
expected_before = reg['initial_overlay']
|
|
else:
|
|
require(dependency['kind'] == 'epoch_delta_after' and dependency['source'].endswith('/delta-after'),
|
|
'Unknown registered overlay dependency')
|
|
previous = tasks_by_id[dependency['source'][:-len('/delta-after')]]
|
|
require(previous['kind'] == 'train' and all(previous[field] == task[field] for field in ('block_id', 'arm'))
|
|
and task_key(previous) + '--verify' in entries, 'Previous overlay is not a committed same-arm training output')
|
|
previous_dir = EXP / 'updates' / task_key(previous)
|
|
expected_before = dict(overlay_file=str(previous_dir / 'delta-after.npz'),
|
|
overlay_manifest=str(previous_dir / 'overlay-manifest.json'))
|
|
require(result['overlay_before'] == expected_before
|
|
and jobs['brain']['overlay_file'] == expected_before['overlay_file']
|
|
and jobs['brain']['overlay_manifest'] == expected_before['overlay_manifest']
|
|
and authority['files']['overlay_before']['path'] == expected_before['overlay_file']
|
|
and authority['files']['overlay_manifest_before']['path'] == expected_before['overlay_manifest'],
|
|
'Actual before overlay differs from registered branch dependency')
|
|
registered_files = {**reg['input_sha256'], **reg['prepared_sha256']}
|
|
for info in authority['files'].values():
|
|
require(jobs['update']['input_sha256'].get(info['path']) == info['sha256'], 'Authority role absent from successful update inputs')
|
|
trusted_sha = registered_files.get(info['path'], tracked.get(info['path'], {}).get('sha256'))
|
|
require(trusted_sha == info['sha256'], 'Authority role lacks registered or committed file identity')
|
|
file_pins[info['path']] = check(info['path'], info['sha256'])
|
|
for name in ('chain-receipt.json', 'update-receipt.json', 'delta-after.npz', 'overlay-manifest.json'):
|
|
path = update_dir / name
|
|
file_pins[str(path)] = check(path, tracked[str(path)])
|
|
chain_sha = file_pins[str(update_dir / 'chain-receipt.json')]['sha256']
|
|
require(jobs['verify']['chain_sha256'] == chain_sha, 'Selected CPU validator used a different chain')
|
|
cpu_path = unlinked(result['cpu_validation']['path'])
|
|
file_pins[str(cpu_path)] = check(cpu_path, tracked[str(cpu_path)])
|
|
require(file_pins[str(cpu_path)]['sha256'] == result['cpu_validation']['sha256'], 'Original CPU receipt pin differs')
|
|
original_validation = read(cpu_path)
|
|
require(original_validation['complete'] is True and original_validation['frozen_U_verify_receipt_recomputed'] is True
|
|
and original_validation['actual_sealed_events_used'] is True
|
|
and original_validation['full_candidate_mapping_verified'] is True
|
|
and original_validation['epoch_authority_sha256'] == authority_sha, 'Original CPU validation incomplete')
|
|
validator = MODEL / 'conditioning-execution-v1/validate_update_cpu.py'
|
|
check(validator, reg['input_sha256'][str(validator)])
|
|
validator_seconds = run_cpu([sys.executable, '-B', str(validator), '--authority', str(authority_path),
|
|
'--authority-sha256', authority_sha, '--update-dir', str(update_dir), '--chain-sha256', chain_sha,
|
|
'--receipt', str(directory / 'fresh-cpu-validation.json')], directory, 'validation')
|
|
fresh = read(directory / 'fresh-cpu-validation.json')
|
|
require(fresh == original_validation, 'Fresh original validator receipt differs from committed receipt')
|
|
import numpy as np
|
|
paths = {role: unlinked(info['path']) for role, info in authority['files'].items()}
|
|
state = load_npz(paths['state'])
|
|
mapping = load_npz(paths['candidate_map'])
|
|
before = load_npz(paths['overlay_before'])
|
|
after = load_npz(update_dir / 'delta-after.npz')
|
|
arrays = dict(step=state['step'], neuron_idx=state['neuron_idx'],
|
|
kc=np.load(paths['kc'], allow_pickle=False), candidate_edge_id=np.load(paths['mask'], allow_pickle=False),
|
|
candidate_sources=mapping['source_neuron_idx'], candidate_targets=mapping['target_neuron_idx'],
|
|
candidate_compartments=mapping['target_compartment_verbatim'], delta_before=before['plastic_delta_mv'],
|
|
base_candidate=np.load(paths['base_candidate'], allow_pickle=False))
|
|
require(len(arrays['kc']) == 4064 and len(arrays['candidate_edge_id']) == 14232, 'Complete identity dimensions differ')
|
|
save_npz(directory / 'inputs.npz', arrays)
|
|
save(directory / 'command.json', command)
|
|
sources = {role: authority['files'][role] for role in
|
|
('conditioning_update_source', 'eligibility_bridge_source', 'phase4j_source')}
|
|
save(directory / 'reference-sources.json', sources)
|
|
reference_seconds = run_cpu([sys.executable, '-B', str(Path(__file__).absolute()),
|
|
'--reference-worker', str(directory)], directory, 'reference')
|
|
reference = load_npz(directory / 'reference.npz')
|
|
receipt = read(directory / 'reference-update-receipt.json')
|
|
stored_receipt = read(update_dir / 'update-receipt.json')
|
|
require(canonical(receipt) == canonical(stored_receipt), 'Original recomputation differs from stored update receipt')
|
|
require(array_evidence(reference['reference_delta_after']) == array_evidence(after['plastic_delta_mv']),
|
|
'Original reference delta bytes differ from committed delta-after')
|
|
for name in OUTPUT_ARRAYS:
|
|
require(array_evidence(reference['reference_' + name]) == stored_receipt['outputs'][name],
|
|
'Reference trace/delta does not match original receipt byte identity')
|
|
fixture = {**arrays, **reference, 'stored_delta_after': after['plastic_delta_mv']}
|
|
save_npz(directory / 'fixture.npz', fixture)
|
|
save(directory / 'authority.json', authority)
|
|
save(directory / 'stored-update-receipt.json', stored_receipt)
|
|
save(directory / 'source-file-pins.json', file_pins)
|
|
timing = read(directory / 'reference-timing.json')
|
|
record = dict(fixture_id=fixture_id, task=task, npz=f'{fixture_id}/fixture.npz',
|
|
command=f'{fixture_id}/command.json', npz_sha256=digest(directory / 'fixture.npz'),
|
|
command_sha256=digest(directory / 'command.json'), authority_sha256=authority_sha,
|
|
original_update_payload_sha256=receipt['payload_sha256'],
|
|
exact_stored_delta_and_trace_verified=True, full_canonical_KC_count=4064, full_candidate_edge_count=14232,
|
|
original_validator_verified_full_graph_mapping=True, reference_timing=timing,
|
|
reference_subprocess_wall_seconds=reference_seconds, original_validator_subprocess_wall_seconds=validator_seconds,
|
|
fixture_end_to_end_seconds=time.perf_counter() - item_started,
|
|
historical_process_seconds={stage: {field: proc[field] for field in ('child_wall_seconds', 'end_to_end_seconds')}
|
|
for stage, proc in process_records.items()},
|
|
input_arrays={name: array_evidence(value) for name, value in arrays.items()},
|
|
output_arrays={name: array_evidence(value) for name, value in reference.items()})
|
|
save(directory / 'fixture.json', record)
|
|
fixtures.append(record)
|
|
print(canonical(dict(fixture=fixture_id, complete=True, task_id=task['task_id'])), flush=True)
|
|
for path, pin in control_pins.items():
|
|
check(path, pin)
|
|
require(not (EXP / '.append-ledger/writer.lock').exists()
|
|
and len(list((EXP / '.append-ledger/entries').iterdir())) == snapshot['entries'],
|
|
'Ledger changed during export; fixture set not published')
|
|
for path, sha in source_pins.items():
|
|
check(path, sha)
|
|
check(Path(__file__).absolute(), own)
|
|
# Recheck all selected immutable input/output pins after the entire export.
|
|
all_pins = {}
|
|
for fixture in fixtures:
|
|
all_pins.update(read(output / fixture['fixture_id'] / 'source-file-pins.json'))
|
|
for path, pin in all_pins.items():
|
|
check(path, pin)
|
|
manifest = dict(schema='conditioning-acceleration-fixtures-v1', complete=True,
|
|
registration_sha256=REG_SHA, source_code_commit=SOURCE_COMMIT,
|
|
exporter_source=dict(path=str(Path(__file__).absolute()), **own),
|
|
created_utc=datetime.now(timezone.utc).isoformat(), selection_sha256=digest(output / 'selection.json'),
|
|
fixtures=fixtures, fixture_count=len(fixtures), coverage=selection,
|
|
total_export_seconds=time.perf_counter() - started,
|
|
original_files_modified=False, brain_executed=False, training_executed=False,
|
|
weights_installed=False, learning_pass_claimed=False,
|
|
authority_limit='committed selected historical processes and original CPU reference only; AR remains incomplete',
|
|
files={str(path.relative_to(output)): identity(path) for path in sorted(output.rglob('*')) if path.is_file()})
|
|
save(output / 'manifest.json', manifest)
|
|
return manifest
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
mode = parser.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument('--output', type=Path)
|
|
mode.add_argument('--reference-worker', type=Path, help=argparse.SUPPRESS)
|
|
args = parser.parse_args()
|
|
if args.reference_worker:
|
|
reference_worker(args.reference_worker)
|
|
else:
|
|
result = export(args.output)
|
|
print(canonical(dict(complete=True, output=str(args.output), fixtures=result['fixture_count'])), flush=True)
|