Digital Drosophila Archive
285 lines
16 KiB
Python
285 lines
16 KiB
Python
"""CPU fixtures and frozen-source checks; never sample or run a brain."""
|
|
from copy import deepcopy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
import boundary_plan as B
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parent.parent
|
|
EVIDENCE = SOURCE.parent / 'reports/phase4ap-publication/evidence'
|
|
if not EVIDENCE.is_dir():
|
|
EVIDENCE = SOURCE.parents[1] / 'registry/phase4ap-conditioning-interface-v2/evidence'
|
|
OUTPUT_ROOT = '/data/digital-drosophila/experiments/boundary-plan-fixture'
|
|
|
|
|
|
def load_source(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class BoundaryPlanTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.template = json.loads((EVIDENCE / 'cases/D1--reference--epoch-001/spec.json').read_text())
|
|
cls.registry = load_source('boundary_test_seed_registry', SOURCE/'olfactory-conditioning-lab-v1/seed_registry.py')
|
|
with patch.dict(sys.modules, {'seed_registry': cls.registry}):
|
|
cls.trial_plan = load_source('boundary_test_trial_plan', SOURCE/'olfactory-conditioning-lab-v1/trial_plan.py')
|
|
|
|
def setUp(self):
|
|
self.plan = B.build_plan()
|
|
self.calls = {c['call_id']: c for c in self.plan['calls']}
|
|
|
|
def spec(self, call):
|
|
return B.derive_case_spec(self.template, call, budget_root=OUTPUT_ROOT)
|
|
|
|
def test_source_protocol_and_template_pins_match_actual_local_bytes(self):
|
|
for path, pin in B.SOURCE_PINS.items():
|
|
self.assertEqual(hashlib.sha256((SOURCE/path).read_bytes()).hexdigest(), pin, path)
|
|
for path, pin in B.AP_ANCHORS.items():
|
|
self.assertEqual(hashlib.sha256((EVIDENCE/path).read_bytes()).hexdigest(), pin, path)
|
|
for arm, pin in B.PARENT_PINS.items():
|
|
raw = (EVIDENCE/f'runs/{arm}--reference--epoch-001/checkpoint-manifest.json').read_bytes()
|
|
self.assertEqual(hashlib.sha256(raw).hexdigest(), pin)
|
|
self.assertEqual(B._sha(self.template), B.TEMPLATE_PAYLOAD_SHA256)
|
|
|
|
def test_exact_inventory_ticks_children_and_no_execution_claim(self):
|
|
self.assertTrue(B.validate_plan(self.plan))
|
|
self.assertEqual(len(self.calls), 23)
|
|
self.assertEqual(sum(c['steps'] for c in self.calls.values()), 125200)
|
|
self.assertEqual(len({c['case_group'] for c in self.calls.values()}), 17)
|
|
self.assertEqual(len(self.plan['imported_parents']), 3)
|
|
self.assertEqual(len(self.plan['cpu_children']), 7)
|
|
self.assertEqual(self.plan['counts']['total_children'], 23+7+1)
|
|
self.assertEqual(self.plan['sampler_child']['uniform_batch_calls'], 2)
|
|
self.assertEqual(self.plan['sampler_child']['execution'], 'GPU_private_generator')
|
|
self.assertIsNone(self.plan['counts']['raw_uniform_draw_count'])
|
|
for key in ('brain_executed', 'inputs_sampled', 'acquisition_executed', 'full_retention_experiment',
|
|
'scientific_learning_claimed', 'external_identity_verified_here', 'execution_authorized'):
|
|
self.assertIs(self.plan[key], False)
|
|
|
|
def test_plan_reconstruction_rejects_mutations_including_booleans(self):
|
|
mutations = [lambda p: p['calls'][0].update(steps=999),
|
|
lambda p: p['counts'].update(new_brain_calls=20),
|
|
lambda p: p['calls'][-1].update(parent='D1--probe--t1--A'),
|
|
lambda p: p['calls'][-1].update(neural_reset=True),
|
|
lambda p: p['calls'][2].update(overlay_before_role='AP.initial_overlay'),
|
|
lambda p: p['mechanical_update'].update(trace_backend='numpy'),
|
|
lambda p: p['source_sha256'].update({'conditioning-interface-v2/PROTOCOL.md': 'a'*64}),
|
|
lambda p: p.update(execution_authorized=0)]
|
|
for mutate in mutations:
|
|
plan = deepcopy(self.plan)
|
|
mutate(plan)
|
|
with self.assertRaises(ValueError):
|
|
B.validate_plan(plan)
|
|
|
|
def test_seeds_are_exact_existing_evaluation_streams_not_training_streams(self):
|
|
streams = B.evaluation_seeds()
|
|
for cue, row in streams.items():
|
|
key = self.trial_plan._input_key(B.BLOCK, 'evaluation', cue, 0, 'base-probe')
|
|
seed, digest = self.registry.seed_for(B.MASTER, 'evaluation', key)
|
|
self.assertEqual(row['key'], list(key))
|
|
self.assertEqual((row['seed'], row['derivation_sha256']), (seed, digest))
|
|
self.assertEqual(row['stream_id'], self.trial_plan._stream_id(B.MASTER, 'evaluation', key))
|
|
train_seed, _ = self.registry.seed_for(B.MASTER, 'training', key)
|
|
self.assertNotEqual(seed, train_seed)
|
|
self.assertGreater(seed, 2**32)
|
|
self.assertLess(seed, 2**63)
|
|
self.assertEqual({self.spec(f'D1--probe--{t}--{cue}')['cs_input_seed'] for t in ('t0','t1','t10')}, {seed})
|
|
self.assertNotEqual(streams['A']['seed'], streams['B']['seed'])
|
|
|
|
def test_plan_and_derivation_never_sample(self):
|
|
with patch('numpy.random.default_rng', side_effect=AssertionError('sampling forbidden')), \
|
|
patch('numpy.random.uniform', side_effect=AssertionError('sampling forbidden')):
|
|
B.build_plan()
|
|
B.evaluation_seeds()
|
|
for call in self.calls:
|
|
self.spec(call)
|
|
|
|
def test_continuous_idle_clock_and_probe_fork_no_feedback(self):
|
|
first, second = self.calls['D1--idle--1s'], self.calls['D1--idle--9s']
|
|
self.assertEqual((first['steps'], first['global_tick_offset'], first['global_next_tick']), (10000,1200,11200))
|
|
self.assertEqual((second['steps'], second['global_tick_offset'], second['global_next_tick']), (90000,11200,101200))
|
|
self.assertEqual(second['parent'], first['call_id'])
|
|
self.assertEqual((first['transition'], second['transition']), ('epoch_update','idle'))
|
|
for t, parent, tick in (('t0','AP.D1.reference.epoch001',1200),
|
|
('t1',first['call_id'],11200), ('t10',second['call_id'],101200)):
|
|
for cue in ('A','B'):
|
|
probe = self.calls[f'D1--probe--{t}--{cue}']
|
|
self.assertEqual((probe['parent'],probe['global_tick_offset'],probe['steps']), (parent,tick,1200))
|
|
self.assertTrue(probe['probe_fork_discard'])
|
|
for key in ('feedback_to_idle','neural_reset','US','plastic_update'):
|
|
self.assertIs(probe[key], False)
|
|
self.assertFalse(any('--probe--' in (c['parent'] or '') for c in self.calls.values()))
|
|
|
|
def test_template_keydiff_preserves_model_APL_delay_and_reset(self):
|
|
saved = deepcopy(self.template)
|
|
for name in self.calls:
|
|
spec = self.spec(name)
|
|
common = set(spec) & set(saved)
|
|
changed = {key for key in common if spec[key] != saved[key]}
|
|
self.assertLessEqual(changed, B.CASE_CHANGED_KEYS | B.STALE_INPUT_KEYS)
|
|
self.assertLessEqual(set(spec)-set(saved), B.CASE_ADDED_KEYS)
|
|
self.assertLessEqual(set(saved)-set(spec), B.STALE_INPUT_KEYS)
|
|
for key in ('model','apl','apl_delay_steps','apl_precision','precision','seed','dt_ms',
|
|
'graph_dir','apl_bundle_dir','coefficients_path','legacy_apl_worker','kind'):
|
|
self.assertEqual(spec[key], saved[key], (name,key))
|
|
self.assertEqual(spec['max_steps'], spec['steps'])
|
|
self.assertEqual(spec['duration_ms'], spec['steps'] * saved['dt_ms'])
|
|
self.assertEqual(self.template, saved)
|
|
|
|
def test_shared_ref_light_oldC_case_groups_produce_identical_specs(self):
|
|
for arm in ('D1','D0','frozen','mechanical'):
|
|
ref = self.spec(f'{arm}--updated--reference')
|
|
self.assertEqual(ref, self.spec(f'{arm}--updated--light'))
|
|
if arm != 'mechanical':
|
|
self.assertEqual(ref['name'], f'AP-{arm}-suffix')
|
|
if arm in ('D0','frozen'):
|
|
self.assertEqual(ref, self.spec(f'{arm}--old-C--reference'))
|
|
|
|
def test_nonolfactory_cases_remove_stale_odor_metadata(self):
|
|
for name, call in self.calls.items():
|
|
spec = self.spec(name)
|
|
if 'cue' not in call:
|
|
self.assertTrue(B.STALE_INPUT_KEYS.isdisjoint(spec), name)
|
|
else:
|
|
for key in ('odor_id','rate_sum_hz','source_panel_row','source_panel_split','target_count','expected_CS_requests'):
|
|
self.assertNotIn(key, spec)
|
|
self.assertEqual(spec['CS_window'], [0,1000])
|
|
|
|
def test_oracle_is_fresh_diagnostic_one_switch_not_restore(self):
|
|
for arm in ('D1','mechanical'):
|
|
row = self.calls[f'{arm}--live-oracle']
|
|
spec = self.spec(row['call_id'])
|
|
self.assertIsNone(row['parent'])
|
|
self.assertEqual(row['steps'], 2200)
|
|
self.assertEqual(row['global_tick_offset'], 0)
|
|
self.assertIs(row['restore_used'], False)
|
|
self.assertIs(spec['within_trial_weight_updates'], True)
|
|
self.assertEqual(spec['diagnostic_boundary_tick'], 1200)
|
|
self.assertIs(spec['scientific_learning_claimed'], False)
|
|
self.assertEqual(self.calls['D1--live-oracle']['overlay_before_role'], 'AP.D1.reference.epoch001.recorded_overlay')
|
|
for name, row in self.calls.items():
|
|
if row['runner'] != 'boundary_oracle':
|
|
self.assertIs(self.spec(name)['within_trial_weight_updates'], False)
|
|
|
|
def test_mechanical_before_is_registration_original_zero_not_D0_output(self):
|
|
registration = json.loads((EVIDENCE/'protocol/preregistered.json').read_text())
|
|
initial = self.plan['initial_overlay']
|
|
self.assertEqual(initial['role'], 'AP.initial_overlay')
|
|
for role in ('overlay_file','overlay_manifest'):
|
|
self.assertEqual(initial[role]['path'], registration['initial_overlay'][role])
|
|
self.assertEqual(initial[role]['sha256'], registration['input_sha256'][initial[role]['path']])
|
|
self.assertEqual(self.calls['mechanical--fresh-parent']['overlay_after_role'], initial['role'])
|
|
self.assertFalse(self.calls['mechanical--fresh-parent']['inherited_learned_delta'])
|
|
update = self.plan['mechanical_update']
|
|
self.assertEqual((update['steps'],update['us_tick'],update['d_command'],update['trace_backend']), (1200,1199,1.0,'torch'))
|
|
self.assertEqual(update['before_overlay_role'], initial['role'])
|
|
|
|
def test_template_drift_unknown_calls_and_bad_output_roots_rejected(self):
|
|
wrong = deepcopy(self.template)
|
|
wrong['model']['threshold_mv'] = -44.0
|
|
with self.assertRaises(ValueError):
|
|
B.derive_case_spec(wrong,'D1--updated--reference',budget_root=OUTPUT_ROOT)
|
|
with self.assertRaises(ValueError):
|
|
self.spec('unregistered-extra-call')
|
|
for root in ('relative','/',B.AP_ROOT,B.AP_ROOT+'/child','/data','/tmp/../test'):
|
|
with self.assertRaises(ValueError):
|
|
B.derive_case_spec(self.template,'D1--updated--reference',budget_root=root)
|
|
|
|
def test_budgets_are_fixed_stop_proposals_not_trial_shortening(self):
|
|
budget = self.plan['budget_proposal']
|
|
self.assertEqual((budget['output_bytes'],budget['free_reserve_bytes']), (4*2**30,10*2**30))
|
|
self.assertEqual((budget['child_timeout_seconds'],budget['workflow_timeout_seconds']), (1800,7200))
|
|
self.assertFalse(budget['approved'])
|
|
self.assertIn('never_shorten_trial_or_change_gate', budget['on_limit'])
|
|
|
|
def test_performance_slot_replays_natural_AP_input_with_full_fresh_cycle(self):
|
|
row = self.calls['performance--natural-light-fresh']
|
|
self.assertEqual((row['diagnostic_mode'],row['steps'],row['global_tick_offset']), ('light',1200,0))
|
|
self.assertIsNone(row['parent'])
|
|
self.assertEqual(row['input_policy'],'replay_AP_D1_epoch001_input')
|
|
self.assertEqual(row['overlay_after_role'],'AP.D1.reference.epoch001.recorded_overlay')
|
|
self.assertFalse(row['new_learning'])
|
|
self.assertFalse(row['input_sampled'])
|
|
update = self.plan['performance_update']
|
|
self.assertEqual((update['us_tick'],update['d_command'],update['trace_backend']), (900,1.0,'torch'))
|
|
self.assertEqual(update['required_regression'],['all_events','final_state','delta_after'])
|
|
names = list(self.calls)
|
|
self.assertLess(names.index(row['call_id']), names.index('mechanical--fresh-parent'))
|
|
|
|
def test_fresh_performance_probe_is_not_retention_continuation(self):
|
|
name = 'performance--evaluation-A-light-fresh'
|
|
row = self.calls[name]
|
|
self.assertEqual((row['runner'],row['steps'],row['global_tick_offset']), ('fresh_probe',1200,0))
|
|
self.assertEqual(row['diagnostic_mode'],'light')
|
|
self.assertIsNone(row['parent'])
|
|
self.assertTrue(row['neural_reset'])
|
|
self.assertEqual(row['overlay_after_role'],'AP.initial_overlay')
|
|
for key in ('US','plastic_update','feedback_to_idle'):
|
|
self.assertIs(row[key],False)
|
|
spec = self.spec(name)
|
|
self.assertEqual(spec['cs_input_seed'],B.evaluation_seeds()['A']['seed'])
|
|
self.assertNotIn('parent_checkpoint_manifest_sha256',spec)
|
|
names = list(self.calls)
|
|
self.assertEqual(names.index(name),names.index('performance--natural-light-fresh')+1)
|
|
self.assertNotEqual(row['case_group'],self.calls['D1--probe--t0--A']['case_group'])
|
|
|
|
|
|
class MechanicalRequestTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.mapping = dict(edge_id=np.arange(B.CANDIDATES,dtype=np.int64),
|
|
source_neuron_idx=np.full(B.CANDIDATES,33,dtype=np.int32),
|
|
target_neuron_idx=np.full(B.CANDIDATES,900,dtype=np.int32),
|
|
target_compartment_verbatim=np.full(B.CANDIDATES,'other',dtype='U8'))
|
|
self.mapping['target_compartment_verbatim'][[4,700]] = B.COMPARTMENT
|
|
self.mapping['source_neuron_idx'][[0,4,700]] = [1,21,7]
|
|
self.dan = np.array([101,102,104,108],dtype=np.int32)
|
|
self.binding = dict(candidate_map_sha256='1'*64,pam15_membership_sha256='2'*64)
|
|
|
|
def run_request(self):
|
|
return B.mechanical_requests(self.mapping,self.dan,identity_binding=self.binding)
|
|
|
|
def test_single_minimum_matching_KC_plus_four_DAN_last_tick(self):
|
|
before = {key:a.copy() for key,a in self.mapping.items()}
|
|
request, audit = self.run_request()
|
|
np.testing.assert_array_equal(request['tick'], np.full(5,1199,dtype=np.int32))
|
|
np.testing.assert_array_equal(request['neuron_idx'], np.array([7,101,102,104,108],dtype=np.int32))
|
|
self.assertEqual(audit['selected_KC_source_neuron_idx'], 7)
|
|
self.assertFalse(audit['external_identity_verified_here'])
|
|
self.assertFalse(audit['accepted_spikes_verified'])
|
|
self.assertFalse(audit['olfactory_input'])
|
|
for key in self.mapping:
|
|
np.testing.assert_array_equal(self.mapping[key],before[key])
|
|
self.dan[0] = 100
|
|
self.assertEqual(request['neuron_idx'][1],101)
|
|
|
|
def test_array_and_identity_schema_mutations_fail_closed(self):
|
|
mutations = [lambda: self.mapping.update(edge_id=self.mapping['edge_id'].astype(np.int32)),
|
|
lambda: self.mapping['edge_id'].__setitem__(1,0),
|
|
lambda: self.mapping.update(source_neuron_idx=self.mapping['source_neuron_idx'][:-1]),
|
|
lambda: self.mapping['source_neuron_idx'].__setitem__(0,-1),
|
|
lambda: self.mapping['target_compartment_verbatim'].__setitem__(slice(None),'y5B2a'),
|
|
lambda: setattr(self,'dan',self.dan.astype(np.int64)),
|
|
lambda: self.dan.__setitem__(1,101),
|
|
lambda: setattr(self,'dan',self.dan[::-1]),
|
|
lambda: self.dan.__setitem__(0,7),
|
|
lambda: self.binding.update(pam15_membership_sha256='not-authenticated')]
|
|
for mutation in mutations:
|
|
self.setUp()
|
|
mutation()
|
|
with self.assertRaises(ValueError):
|
|
self.run_request()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|