Digital Drosophila Archive
475 lines
29 KiB
Python
475 lines
29 KiB
Python
"""Finite AO runner: exact frozen brain, two fresh epochs, delta-only carry.
|
|
|
|
Never starts the full conditioning protocol. Partial rows are preserved. Each
|
|
runtime is sealed immediately; coordinator evidence lives outside its seal.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import traceback
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
|
|
import conditioning_preflight_prepare as P
|
|
import conditioning_preflight_inputs as I
|
|
import conditioning_inputs as IN
|
|
import conditioning_update as U
|
|
import conditioning_epoch as E
|
|
import conditioning_preflight_compare as Q
|
|
import engineering_compare as COMP
|
|
import engineering_workflow as ENV
|
|
import runtime_light as R
|
|
import runtime_continuation as C
|
|
|
|
EXP, MODEL, ROOT = P.EXP, P.MODEL, P.ROOT
|
|
require, read, save, digest = P.require, P.read, P.save, P.digest
|
|
|
|
|
|
def npz(path):
|
|
with np.load(path, allow_pickle=False) as z:
|
|
return {k: z[k].copy() for k in z.files}
|
|
|
|
|
|
def row_by(reg, name):
|
|
matches = [r for r in reg['plan']['calls'] if r['row_id'] == name]
|
|
require(len(matches) == 1, 'Only an exact registered row may execute')
|
|
return matches[0]
|
|
|
|
|
|
def locations(name):
|
|
require(type(name) is str and name and all(c.isalnum() or c in '-_' for c in name), 'Unsafe row')
|
|
return dict(run=EXP / 'runs' / name, update=EXP / 'updates' / name,
|
|
runtime=EXP / 'receipts' / (name + '.runtime.json'), execution=EXP / 'receipts' / (name + '.process.json'),
|
|
authority=EXP / 'receipts' / (name + '.authority.json'),
|
|
stdout=EXP / 'logs' / (name + '.stdout.log'), stderr=EXP / 'logs' / (name + '.stderr.log'))
|
|
|
|
|
|
def budget(reg):
|
|
files = list(EXP.rglob('*'))
|
|
require(not any(p.is_symlink() for p in files), 'Linked output is forbidden')
|
|
used = sum(p.stat().st_size for p in files if p.is_file())
|
|
free = shutil.disk_usage(EXP).free
|
|
b = reg['budget']
|
|
require(used <= b['output_budget_bytes'] and free >= b['minimum_free_reserve_bytes'] + max(0, b['output_budget_bytes'] - used),
|
|
'Resource cap/reserve exceeded; preserve partials')
|
|
return dict(used_bytes=used, free_bytes=free, **b)
|
|
|
|
|
|
def tables():
|
|
metadata = read(EXP / 'inputs/static/rate-tables.json')
|
|
return {cue: dict(table, **npz(EXP / f'inputs/static/{cue}-rates.npz')) for cue, table in metadata.items()}
|
|
|
|
|
|
def original_sampler_nodes(raw):
|
|
"""Only the four original L sampling statements, no old brain import."""
|
|
require(__import__('hashlib').sha256(raw).hexdigest() == I.SOURCE_PINS[I.SOURCE_REL], 'L source changed')
|
|
methods = [n for n in ast.walk(ast.parse(raw)) if isinstance(n, ast.FunctionDef) and n.name == 'run_rates']
|
|
require(len(methods) == 1, 'Original sampling method is not unique')
|
|
chosen = []
|
|
for node in methods[0].body:
|
|
if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id in ('rg', 'gen', 'forced') for t in node.targets):
|
|
chosen.append(node)
|
|
elif isinstance(node, ast.Expr) and isinstance(node.value, ast.Call) and ast.unparse(node.value.func) == 'gen.manual_seed':
|
|
chosen.append(node)
|
|
require(len(chosen) == 4 and [ast.unparse(n).split('=')[0].strip() for n in chosen[:2]] == ['rg', 'gen'],
|
|
'Original sampler AST structure changed')
|
|
return chosen
|
|
|
|
|
|
def sampler_check(reg, table, generated):
|
|
import torch
|
|
seed = reg['generation']['highbit_probe_seed']
|
|
nodes = original_sampler_nodes((ROOT / I.SOURCE_REL).read_bytes())
|
|
env = dict(torch=torch, targets=table['targets'], rates_hz=table['rates_hz'], seed=seed, steps=1200,
|
|
self=SimpleNamespace(dt=.1, device=torch.device('cuda:0')))
|
|
exec(compile(ast.Module(body=nodes, type_ignores=[]), str(ROOT / I.SOURCE_REL), 'exec'), env)
|
|
original = IN.requests_from_forced(env['forced'].cpu().numpy(), table['targets'], cs_start_tick=0, cs_stop_tick=1000)
|
|
require(all(COMP.bit_equal(original['requests'][k], generated['cs_requests'][k]) for k in ('tick', 'neuron_idx')),
|
|
'Actual CUDA original-L sampling expression disagrees')
|
|
identities = []
|
|
for value in (seed, seed & 0xffffffff):
|
|
gen = torch.Generator(device='cuda:0').manual_seed(value)
|
|
matrix = torch.rand((1200, len(table['targets'])), generator=gen, device='cuda:0')
|
|
identities.append(IN.array_identity(matrix.cpu().numpy()))
|
|
require(identities[0] != identities[1], 'Actual CUDA full63bit and low32bit sequence identical; stop, no seed substitution')
|
|
return dict(original_L_sampler_AST_sha256=__import__('hashlib').sha256(ast.dump(ast.Module(body=nodes, type_ignores=[]), include_attributes=False).encode()).hexdigest(),
|
|
exact_sparse_CS_requests=True, full_seed=seed, low32_probe_seed=seed & 0xffffffff,
|
|
uniforms_identities=identities, highbit_influence_observed=True,
|
|
additional_probe_matrices=3, probe_shape=[1200, len(table['targets'])],
|
|
new_input_seed_selected=False, old_full_trial_equivalence_claimed=False)
|
|
|
|
|
|
def materialize(reg_sha):
|
|
reg = P.read_registration(reg_sha)
|
|
admission()
|
|
ENV.preflight(); budget(reg)
|
|
target = EXP / 'inputs/generated'
|
|
require(not target.exists(), 'Never rerun or overwrite actual input generation')
|
|
target.mkdir()
|
|
rate = tables()
|
|
dan = npz(EXP / 'inputs/static/populations.npz')[reg['plan']['dan_type']]
|
|
records = []
|
|
for index, epoch in enumerate(reg['plan']['selected_epochs']):
|
|
started = time.perf_counter()
|
|
value = I.generate_epoch_inputs(epoch, rate, dan, device=reg['generation']['device'],
|
|
max_uniform_draws=reg['budget']['max_uniform_draws'], conflict_policy=reg['generation']['conflict_policy'])
|
|
directory = target / f'epoch-{index:03d}'
|
|
directory.mkdir()
|
|
for kind in ('requests', 'cs_requests', 'us_requests'):
|
|
P.save_npz(directory / (kind + '.npz'), **value[kind])
|
|
P.save_npz(directory / 'source-mask.npz', source_mask=value['source_mask'])
|
|
save(directory / 'sampler-receipt.json', value['audit'])
|
|
if index == 0:
|
|
save(target / 'actual-CUDA-sampler-check.json', sampler_check(reg, rate[epoch['cs']['cue']], value))
|
|
records.append(dict(index=index, epoch_id=epoch['epoch_id'], wall_seconds=time.perf_counter() - started,
|
|
CS_requests=len(value['cs_requests']['tick']), US_requests=len(value['us_requests']['tick'])))
|
|
for row in reg['plan']['calls']:
|
|
src = target / f"epoch-{row['input_epoch_index']:03d}" / 'requests.npz'
|
|
dst = Path(reg['cases'][row['row_id']]) / 'requests.npz'
|
|
with src.open('rb') as inp, dst.open('xb') as out:
|
|
shutil.copyfileobj(inp, out)
|
|
require(digest(src) == digest(dst), 'Shared input copy differs')
|
|
pins = {str(p): digest(p) for p in target.rglob('*') if p.is_file()}
|
|
pins.update({str(Path(path) / 'requests.npz'): digest(Path(path) / 'requests.npz') for path in reg['cases'].values()})
|
|
P.read_registration(reg_sha)
|
|
save(EXP / 'receipts/input-generation.json', dict(complete=True, registration_sha256=reg_sha,
|
|
inputs_sha256=pins, rows=records, GPU_UUID=ENV.GPU, full63bit_delivery_and_influence_checked=True,
|
|
input_generation_calls=2, brain_runs=0, parameters_changed=False))
|
|
|
|
|
|
def generated_inputs(reg_sha):
|
|
receipt = read(EXP / 'receipts/input-generation.json')
|
|
require(receipt.get('complete') is True and receipt.get('registration_sha256') == reg_sha and receipt.get('brain_runs') == 0,
|
|
'Generated inputs are not bound to registration')
|
|
process = read(EXP / 'receipts/input-generation.process.json')
|
|
require(process['exit_code'] == 0 and process['registration_sha256'] == reg_sha and
|
|
process['receipt_sha256'] == digest(EXP / 'receipts/input-generation.json'), 'Input generation process missing')
|
|
for k in ('stdout', 'stderr'):
|
|
require(process[k + '_sha256'] == digest(EXP / 'logs' / ('input-generation.' + k + '.log')), 'Input generation log changed')
|
|
P.verify_pins(receipt['inputs_sha256'])
|
|
return receipt
|
|
|
|
|
|
def verify_row(reg, reg_sha, row):
|
|
paths = locations(row['row_id'])
|
|
record = read(paths['runtime']); process = read(paths['execution'])
|
|
require(record['complete'] is True and record['registration_sha256'] == reg_sha and P.same(record['row'], row),
|
|
'Runtime record differs from registered row')
|
|
require(process['exit_code'] == 0 and process['registration_sha256'] == reg_sha and
|
|
process['runtime_sha256'] == digest(paths['runtime']), 'Runtime process did not complete exactly')
|
|
for key in ('stdout', 'stderr'):
|
|
require(process[key + '_sha256'] == digest(paths[key]), 'Process log changed')
|
|
P.verify_pins(record['files_sha256'])
|
|
checkpoint = C.load_checkpoint(paths['run'], record['checkpoint_manifest_sha256'])
|
|
require(checkpoint['fresh'] and checkpoint['steps'] == row['steps'] and checkpoint['global_tick_offset'] == 0,
|
|
'Normal epoch must be fresh, not a continuation')
|
|
ENV.verify_environment(record['environment'], checkpoint['summary']['apl_component']['environment'])
|
|
chain = read(paths['update'] / 'chain-receipt.json')
|
|
require(chain['authority_sha256'] == digest(paths['authority']) and
|
|
chain['source_neural_checkpoint_manifest_sha256'] == record['checkpoint_manifest_sha256'] and
|
|
chain['epoch_id'] == row['epoch_id'], 'Delta does not derive from this actual sealed neural epoch')
|
|
P.verify_pins(chain['input_files_sha256'])
|
|
for name, sha in chain['output_files_sha256'].items():
|
|
require(digest(paths['update'] / name) == sha, 'Updated snapshot changed')
|
|
return record
|
|
|
|
|
|
def before_overlay(reg, reg_sha, row):
|
|
previous = row['previous_delta_row_id']
|
|
if previous is None:
|
|
return reg['initial_overlay']
|
|
parent = row_by(reg, previous)
|
|
require(parent['arm'] == row['arm'] and parent['diagnostic_mode'] == row['diagnostic_mode'] and
|
|
parent['input_epoch_index'] + 1 == row['input_epoch_index'], 'Invalid cross-epoch delta link')
|
|
verify_row(reg, reg_sha, parent)
|
|
directory = locations(previous)['update']
|
|
return dict(overlay_file=str(directory / 'delta-after.npz'), overlay_manifest=str(directory / 'overlay-manifest.json'))
|
|
|
|
|
|
def make_authority(reg, row, checkpoint_sha, overlay):
|
|
paths = locations(row['row_id']); case = Path(reg['cases'][row['row_id']]); spec = read(case / 'spec.json')
|
|
metadata = read(overlay['overlay_manifest'])
|
|
role_paths = dict(spec=case / 'spec.json', requests=case / 'requests.npz',
|
|
overlay_before=overlay['overlay_file'], overlay_manifest_before=overlay['overlay_manifest'],
|
|
state=paths['run'] / 'state.npz', kc=Path(spec['apl_bundle_dir']) / 'kc_globals.npy',
|
|
canonical_kc=Path(spec['apl_bundle_dir']) / 'kc_globals.npy', candidate_map=EXP / 'inputs/static/candidate-map.npz',
|
|
mask=metadata['mask_path'], base_candidate=metadata['base_candidate_path'],
|
|
phase4j_source=reg['phase4j_source'], eligibility_bridge_source=reg['bridge_source'],
|
|
conditioning_update_source=MODEL / 'conditioning_update.py', runtime_light_source=MODEL / 'runtime_light.py',
|
|
runtime_continuation_source=MODEL / 'runtime_continuation.py', am_overlay_source=metadata['am_overlay_path'])
|
|
epoch = reg['plan']['selected_epochs'][row['input_epoch_index']]
|
|
return dict(schema=E.SCHEMA, epoch_id=row['epoch_id'], diagnostic_mode=row['diagnostic_mode'],
|
|
command=dict(steps=row['steps'], compartment=reg['plan']['compartment_verbatim'],
|
|
us_tick=epoch['us']['tick'], d_command=row['d_command'], frozen=row['frozen'], trace_backend=reg['generation']['trace_backend']),
|
|
checkpoint_dir=str(paths['run']), checkpoint_manifest_sha256=checkpoint_sha,
|
|
files={role: dict(path=str(path), sha256=digest(path)) for role, path in role_paths.items()},
|
|
frozen_inputs={k: v for k, v in reg['input_sha256'].items()
|
|
if k not in (str(P.AN / 'protocol/preregistered.json'), str(P.AN / 'manifest.json'), str(P.AN / 'audit/final-status.json'))})
|
|
|
|
|
|
def replay_update(authority, receipt):
|
|
files = {k: Path(v['path']) for k, v in authority['files'].items()}
|
|
state = npz(files['state']); mapping = npz(files['candidate_map']); old = npz(files['overlay_before'])
|
|
command = authority['command']
|
|
return U.verify_receipt(receipt, step=state['step'], neuron_idx=state['neuron_idx'],
|
|
kc=np.load(files['kc'], allow_pickle=False), candidate_edge_id=mapping['edge_id'],
|
|
candidate_sources=mapping['source_neuron_idx'], candidate_targets=mapping['target_neuron_idx'],
|
|
candidate_compartments=mapping['target_compartment_verbatim'], delta_before=old['plastic_delta_mv'],
|
|
base_candidate=np.load(files['base_candidate'], allow_pickle=False), steps=command['steps'], n_neurons=166700,
|
|
compartment=command['compartment'], us_tick=command['us_tick'], d_command=command['d_command'],
|
|
frozen=command['frozen'], trace_backend=command['trace_backend'],
|
|
bridge_source=files['eligibility_bridge_source'], phase4j_source=files['phase4j_source'])
|
|
|
|
|
|
def worker(reg_sha, row_id):
|
|
reg = P.read_registration(reg_sha); generated_inputs(reg_sha)
|
|
admission()
|
|
row = row_by(reg, row_id); paths = locations(row_id)
|
|
require(not any(paths[k].exists() for k in ('run', 'update', 'runtime', 'authority')), 'Partial row cannot be retried')
|
|
ENV.preflight(); before = budget(reg)
|
|
overlay = before_overlay(reg, reg_sha, row)
|
|
case = Path(reg['cases'][row_id])
|
|
protected = {Path(p).parent for p in (*reg['input_sha256'], *reg['source_sha256'], *reg['prepared_file_sha256'])}
|
|
protected.update({EXP / 'inputs', EXP / 'updates'})
|
|
# Build/run siblings must never touch any current or earlier learned input.
|
|
R.execution_paths(case, paths['run'], EXP / 'extension-build', protected=protected)
|
|
started = time.perf_counter(); started_utc = datetime.now(timezone.utc).isoformat()
|
|
execute = R.run_reference if row['diagnostic_mode'] == 'reference' else R.run_light
|
|
execute(case, paths['run'], EXP / 'extension-build', overlay['overlay_file'], overlay['overlay_manifest'])
|
|
checkpoint_sha = C.seal_checkpoint(paths['run'])['sha256']
|
|
runtime_seconds = time.perf_counter() - started
|
|
authority = make_authority(reg, row, checkpoint_sha, overlay)
|
|
save(paths['authority'], authority)
|
|
update_started = time.perf_counter()
|
|
E.postprocess_epoch(paths['authority'], digest(paths['authority']), paths['update'])
|
|
receipt = read(paths['update'] / 'update-receipt.json')
|
|
replay_update(authority, receipt)
|
|
delta_before = npz(overlay['overlay_file'])['plastic_delta_mv']
|
|
_, after, _, _ = R.load_overlay(paths['update'] / 'delta-after.npz', paths['update'] / 'overlay-manifest.json')
|
|
binding = Q.audit_delta_binding(delta_before, after['plastic_delta_mv'], after['plastic_delta_mv'],
|
|
update_receipt=receipt, parent_snapshot_sha256=digest(overlay['overlay_file']),
|
|
child_manifest=read(paths['update'] / 'overlay-manifest.json'))
|
|
require(binding['pass_all'], 'Actual recomputed delta/child binding failed')
|
|
save(EXP / 'receipts' / (row_id + '.binding.json'), binding)
|
|
record = dict(complete=True, registration_sha256=reg_sha, row=row,
|
|
checkpoint_manifest_sha256=checkpoint_sha, environment=ENV.runtime_environment(),
|
|
runtime_and_immediate_seal_wall_seconds=runtime_seconds,
|
|
postprocess_and_recompute_wall_seconds=time.perf_counter() - update_started,
|
|
started_utc=started_utc, budget_before=before, budget_after=budget(reg),
|
|
files_sha256={str(p): digest(p) for p in (paths['authority'], *paths['update'].iterdir(),
|
|
EXP / 'receipts' / (row_id + '.binding.json'))},
|
|
before_overlay_file_sha256=digest(overlay['overlay_file']), epoch_update_recomputed=True,
|
|
learning_success_claimed=False, full_conditioning_executed=False)
|
|
ENV.verify_environment(record['environment'], read(paths['run'] / 'summary.json')['apl_component']['environment'])
|
|
P.read_registration(reg_sha)
|
|
save(paths['runtime'], record)
|
|
print('COMPLETE', row_id, receipt['counts']['actual_bitwise_changed_edges'], flush=True)
|
|
|
|
|
|
def child_process(reg, reg_sha, action, row_id=None):
|
|
if row_id is None:
|
|
out, err = [EXP / 'logs' / ('input-generation.' + k + '.log') for k in ('stdout', 'stderr')]
|
|
else:
|
|
p = locations(row_id); out, err = p['stdout'], p['stderr']
|
|
require(not out.exists() and not err.exists(), 'Partial child logs exist; never retry')
|
|
command = [sys.executable, '-B', str(Path(__file__)), action, '--registration-sha256', reg_sha]
|
|
if row_id is not None: command += ['--row-id', row_id]
|
|
started = time.perf_counter()
|
|
with out.open('xb') as stdout, err.open('xb') as stderr:
|
|
result = subprocess.run(command, cwd=MODEL, stdout=stdout, stderr=stderr,
|
|
timeout=reg['budget']['child_timeout_seconds'])
|
|
require(result.returncode == 0, 'Child failed; preserve partial files and inspect logs')
|
|
receipt = dict(exit_code=result.returncode, registration_sha256=reg_sha,
|
|
stdout_sha256=digest(out), stderr_sha256=digest(err), wall_seconds=time.perf_counter() - started)
|
|
if row_id is None:
|
|
receipt['receipt_sha256'] = digest(EXP / 'receipts/input-generation.json')
|
|
save(EXP / 'receipts/input-generation.process.json', receipt)
|
|
else:
|
|
receipt['runtime_sha256'] = digest(locations(row_id)['runtime'])
|
|
save(locations(row_id)['execution'], receipt)
|
|
|
|
|
|
def run(reg_sha):
|
|
reg = P.read_registration(reg_sha)
|
|
admission()
|
|
completed = []
|
|
try:
|
|
child_process(reg, reg_sha, 'materialize')
|
|
generated_inputs(reg_sha)
|
|
for row in reg['plan']['calls']:
|
|
print('START', row['row_id'], flush=True)
|
|
budget(reg)
|
|
child_process(reg, reg_sha, 'worker', row['row_id'])
|
|
verify_row(reg, reg_sha, row)
|
|
completed.append(row['row_id'])
|
|
print('COMPLETE', row['row_id'], flush=True)
|
|
save(EXP / 'audit/execution-complete.json', dict(registration_sha256=reg_sha, complete=True,
|
|
completed=completed, calls=12, ticks=14400, learning_success_claimed=False))
|
|
except Exception as error:
|
|
save(EXP / 'audit/stopped.json', dict(complete=False, registration_sha256=reg_sha,
|
|
error_type=type(error).__name__, error=str(error), traceback=traceback.format_exc(),
|
|
completed_rows=completed, partials_preserved=True, no_automatic_retry=True))
|
|
raise
|
|
|
|
|
|
def admission():
|
|
require(not (EXP / 'audit/stopped.json').exists() and not (EXP / 'audit/execution-complete.json').exists() and
|
|
not (EXP / 'manifest.json').exists(), 'Stopped/completed protocol cannot be automatically restarted')
|
|
|
|
|
|
def expected_weight_hash(reg, reg_sha, row):
|
|
"""Independent CPU expected installation from authorized base and delta."""
|
|
overlay = before_overlay(reg, reg_sha, row)
|
|
helper, plan, _, _ = R.load_overlay(overlay['overlay_file'], overlay['overlay_manifest'])
|
|
spec = read(Path(reg['cases'][row['row_id']]) / 'spec.json')
|
|
graph = Path(spec['graph_dir']); arrays = {}
|
|
try:
|
|
for name in ('nodes', 'weights', 'edge_id', 'src'):
|
|
arrays[name] = np.load(graph / (name + '.npy'), mmap_mode='r', allow_pickle=False)
|
|
require(np.array_equal(arrays['nodes'], np.arange(166700, dtype=np.int32)), 'Global graph node identity changed')
|
|
apl = np.load(Path(spec['apl_bundle_dir']) / 'apl_globals.npy', allow_pickle=False)
|
|
mask = np.isin(arrays['src'], apl)
|
|
require(int(mask.sum()) == 5845, 'Original fast APL mask differs')
|
|
mapping = helper.map_installation(arrays['weights'], arrays['edge_id'], mask, plan)
|
|
require(mapping['missing_candidate_count'] == 0, 'Full candidate mask not installed')
|
|
base_sha, effective_sha = helper.expected_hashes(arrays['weights'], mask, mapping)
|
|
return dict(base_raw_sha256=base_sha, effective_raw_sha256=effective_sha,
|
|
expected_from_base_and_overlay_not_run_summary=True)
|
|
finally:
|
|
for a in arrays.values(): a._mmap.close()
|
|
|
|
|
|
def analyze(reg_sha):
|
|
reg = P.read_registration(reg_sha); generated_inputs(reg_sha)
|
|
require(not (EXP / 'audit/stopped.json').exists(), 'Stopped protocol has no PASS analysis')
|
|
complete = read(EXP / 'audit/execution-complete.json')
|
|
require(complete['complete'] and complete['registration_sha256'] == reg_sha and
|
|
complete['completed'] == [r['row_id'] for r in reg['plan']['calls']], 'Missing finite calls')
|
|
records = {r['row_id']: verify_row(reg, reg_sha, r) for r in reg['plan']['calls']}
|
|
pairs, summaries, deltas, cross_arm = [], [], {}, []
|
|
for row in reg['plan']['calls']:
|
|
paths = locations(row['row_id']); authority = read(paths['authority'])
|
|
require(P.same(authority, make_authority(reg, row, records[row['row_id']]['checkpoint_manifest_sha256'],
|
|
before_overlay(reg, reg_sha, row))), 'Authority differs from registered row and actual parent')
|
|
update = read(paths['update'] / 'update-receipt.json')
|
|
replay_update(authority, update)
|
|
delta = npz(paths['update'] / 'delta-after.npz')['plastic_delta_mv']
|
|
deltas[row['row_id']] = delta
|
|
if row['arm'] in ('D0', 'frozen'):
|
|
require(not np.any(delta.view(np.uint32)), 'D0/frozen must remain exact positive zero')
|
|
summary = read(paths['run'] / 'summary.json')
|
|
population = npz(EXP / 'inputs/static/populations.npz')
|
|
state = npz(paths['run'] / 'state.npz')
|
|
run_record = COMP.load_run(paths['run'])
|
|
try:
|
|
us = npz(EXP / f"inputs/generated/epoch-{row['input_epoch_index']:03d}/us_requests.npz")
|
|
us_keys = us['tick'].astype(np.int64) * 166700 + us['neuron_idx']
|
|
request_keys = run_record['requests'][:, 0] * 166700 + run_record['requests'][:, 1]
|
|
dan_requests = run_record['requests'][np.isin(request_keys, us_keys)]
|
|
require(len(dan_requests) == len(us_keys), 'Scheduled US requests not fully accounted')
|
|
delivery = dict(scheduled_D=row['d_command'], requested=int(len(dan_requests)),
|
|
accepted=int(dan_requests[:, 3].sum()), rejected=int(dan_requests[:, 4].sum()),
|
|
actual_DAN_spikes=int(state['counts'][population[reg['plan']['dan_type']]].sum()),
|
|
actual_DAN_spikes_gate_update=False, artificial_US_not_dopamine_concentration=True)
|
|
finally:
|
|
COMP.close_run(run_record)
|
|
summaries.append(dict(row_id=row['row_id'], total_spikes=summary['total_spikes'],
|
|
active_neurons=int(np.count_nonzero(state['counts'])), all_finite=summary['all_finite'],
|
|
update_counts=update['counts'], DAN_delivery=delivery,
|
|
runtime_seconds=records[row['row_id']]['runtime_and_immediate_seal_wall_seconds'],
|
|
effective_weight_sha256=summary['effective_weights_raw_sha256']))
|
|
for arm in ('D1', 'D0', 'frozen'):
|
|
for epoch in range(2):
|
|
names = [f'{arm}--{mode}--epoch-{epoch:03d}' for mode in ('reference', 'light')]
|
|
a, b = [COMP.load_run(locations(n)['run']) for n in names]
|
|
try:
|
|
requests = npz(EXP / f'inputs/generated/epoch-{epoch:03d}/requests.npz')
|
|
expected = expected_weight_hash(reg, reg_sha, row_by(reg, names[0]))
|
|
pair = Q.compare_epoch_pair(a, b, expected_requests=np.column_stack((requests['tick'], requests['neuron_idx'])),
|
|
expected_steps=1200, expected_effective_weight_sha256=expected['effective_raw_sha256'])
|
|
require(pair['pass_all'] and COMP.bit_equal(deltas[names[0]], deltas[names[1]]), 'Reference/light exact regression failed')
|
|
pairs.append(dict(arm=arm, epoch=epoch, exact_delta_after=True, expected_weights=expected, comparison=pair))
|
|
finally:
|
|
COMP.close_run(a); COMP.close_run(b)
|
|
for mode in ('reference', 'light'):
|
|
for left_arm, right_arm, epoch in (('D0', 'frozen', 0), ('D0', 'frozen', 1), ('D1', 'D0', 0)):
|
|
left = COMP.load_run(locations(f'{left_arm}--{mode}--epoch-{epoch:03d}')['run'])
|
|
right = COMP.load_run(locations(f'{right_arm}--{mode}--epoch-{epoch:03d}')['run'])
|
|
try:
|
|
comparison = COMP.compare_fresh(left, right)
|
|
require(comparison['numerical_exact'], 'Identical-input/zero-delta cross-arm neural behavior differs')
|
|
cross_arm.append(dict(left=left_arm, right=right_arm, mode=mode, epoch=epoch, comparison=comparison))
|
|
finally:
|
|
COMP.close_run(left); COMP.close_run(right)
|
|
d1_rows = [s for s in summaries if s['row_id'].startswith('D1--')]
|
|
nonzero = all(s['update_counts']['actual_bitwise_changed_edges'] > 0 for s in d1_rows)
|
|
installed = {}
|
|
for mode in ('reference', 'light'):
|
|
first, second = (locations(f'D1--{mode}--epoch-{j:03d}') for j in range(2))
|
|
raw = R._raw(deltas[f'D1--{mode}--epoch-000'])
|
|
installation = read(second['run'] / 'overlay-audit.json')
|
|
require(installation['delta_raw_sha256'] == raw and
|
|
digest(first['update'] / 'delta-after.npz') == records[f'D1--{mode}--epoch-001']['before_overlay_file_sha256'],
|
|
'Epoch 2 did not install epoch 1 learned delta')
|
|
installed[mode] = dict(exact=True, nonzero_delta=int(np.count_nonzero(deltas[f'D1--{mode}--epoch-000'])))
|
|
status = 'PASS-EXACT-TWO-EPOCH-DELTA-INTERFACE-NOT-LEARNING' if nonzero else 'INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE'
|
|
return dict(status=status, registration_sha256=reg_sha, calls=12, ticks=14400, pairs=pairs, cross_arm=cross_arm,
|
|
rows=summaries, epoch2_installations=installed, nonzero_D1_update_coverage=nonzero,
|
|
D0_and_frozen_exact_zero=True, complete_conditioning_executed=False, learned_candidate=False,
|
|
learning_success_claimed=False, retention_transition_supported=False,
|
|
scope='Two selected A epochs, one training replicate, one PAM15 compartment; no acquisition or A/B behavioral test; no general dynamics or learning claim',
|
|
light_intermediate_states_observed=False)
|
|
|
|
|
|
def finalize(reg_sha):
|
|
require(not (EXP / 'manifest.json').exists() and not (EXP / 'audit/final-status.json').exists(), 'Do not overwrite final/partial seal')
|
|
report = analyze(reg_sha)
|
|
save(EXP / 'audit/interface-analysis.json', report)
|
|
an = P.P.read_registration(P.AN_REG)
|
|
helper = P.P.load_module('phase4ao_final_seal_helper', ROOT / P.P.AI_HELPER_REL, P.P.AI_HELPER_SHA)
|
|
save(EXP / 'regression/AN-seal-after.json', helper.verify_seal(P.AN))
|
|
save(EXP / 'regression/AM-seals-after.json', P.P.verify_authority(an['anchors']))
|
|
old = P.P.verify_old_integrity(an)
|
|
save(EXP / 'regression/old-integrity-after.json', old)
|
|
P.read_registration(reg_sha)
|
|
final = dict(status=report['status'], analysis_sha256=digest(EXP / 'audit/interface-analysis.json'),
|
|
registration_sha256=reg_sha, old_integrity_pass=True, old_roots=len(old),
|
|
prior_AN_AM_seals_preserved=True, frozen_21_and_previous_draft_20_unchanged=True,
|
|
five_reference_trials='Previously sealed AM evidence retained, not rerun in this finite interface gate',
|
|
learned_candidate=False, learning_success_claimed=False, full_conditioning_executed=False,
|
|
model_directory_sealed=False, new_experiment_only=True, budget=budget(P.read_registration(reg_sha)))
|
|
save(EXP / 'audit/final-status.json', final)
|
|
paths = sorted(p for p in EXP.rglob('*') if p.is_file())
|
|
require(not any(p.is_symlink() for p in EXP.rglob('*')) and not any('__pycache__' in p.parts for p in paths), 'Unsafe seal payload')
|
|
rows = [dict(path=p.relative_to(EXP).as_posix(), bytes=p.stat().st_size, sha256=digest(p)) for p in paths]
|
|
save(EXP / 'manifest.json', dict(root=str(EXP), files=rows))
|
|
with (EXP / 'checksums.sha256').open('x', encoding='utf-8', newline='\n') as out:
|
|
out.writelines(r['sha256'] + ' ' + r['path'] + '\n' for r in rows)
|
|
out.write(digest(EXP / 'manifest.json') + ' manifest.json\n')
|
|
helper.verify_seal(EXP)
|
|
print(json.dumps(dict(status=report['status'], manifest_sha256=digest(EXP / 'manifest.json'),
|
|
final_sha256=digest(EXP / 'audit/final-status.json')), sort_keys=True), flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
require(sys.dont_write_bytecode, 'Use scientific Python -B')
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('action', choices=('materialize', 'worker', 'run', 'finalize'))
|
|
parser.add_argument('--registration-sha256', required=True)
|
|
parser.add_argument('--row-id')
|
|
args = parser.parse_args()
|
|
if args.action == 'worker': worker(args.registration_sha256, args.row_id)
|
|
elif args.action == 'materialize': materialize(args.registration_sha256)
|
|
elif args.action == 'run': run(args.registration_sha256)
|
|
else: finalize(args.registration_sha256)
|