Digital Drosophila Archive
250 lines
17 KiB
Python
250 lines
17 KiB
Python
"""Pure, finite boundary-validation proposal; no disk, sampling, runtime or GPU.
|
|
|
|
Pins and claimed array identities are descriptors, NOT verification of disk bytes,
|
|
biological membership, registration or execution permission. The coordinator must
|
|
authenticate them and safely materialize cases. This is not acquisition, a full
|
|
retention experiment, or evidence of learning. No thresholds/model knobs change.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
import hashlib
|
|
import json
|
|
from pathlib import PurePosixPath
|
|
|
|
import numpy as np
|
|
|
|
|
|
SCHEMA = 'conditioning-boundary-plan-v1-draft'
|
|
MASTER = 20260930
|
|
BLOCK = 'physiological__PAM15__reinforce_A__rep01'
|
|
COMPARTMENT = "y5B'2a"
|
|
N, CANDIDATES = 166700, 14232
|
|
AP_ROOT = '/data/digital-drosophila/experiments/phase4ap-conditioning-interface-v2'
|
|
TEMPLATE_PAYLOAD_SHA256 = 'fbd892d6b05e93747f70b70051b36b43922b773633de9ad64bb00af6206a51e1'
|
|
SOURCE_PINS = {
|
|
'conditioning-interface-v2/coordinator.py': '7862fed19bcf3878f42783af6286574371a2d58c66719c99cabab8f18ec99a34',
|
|
'conditioning-interface-v2/isolated_workers.py': 'e47651a8ea99d7c44115107ff6b5dc71312183609ae11c891db65bb21ed61e30',
|
|
'conditioning-interface-v2/test_coordinator.py': 'ffd2951c2bf23ea1a4a8d598fa1be9f07f71e60ad28fe5a06eed17537946d900',
|
|
'conditioning-interface-v2/test_isolated_workers.py': 'd72fcc10f32a1284fa49491cb19f29e83b37b472051985e2fda56799c14a2e60',
|
|
'conditioning-interface-v2/PROTOCOL.md': '2a1c553e6be87df02db8d357f00eb88c9dea3c9b75b6ef670dd55f45492d7518',
|
|
'olfactory-conditioning-lab-v1/seed_registry.py': '0da033b132095ec3087b5d9f64e53c2e391fa50ab0b7653b31296eeb84799b7e',
|
|
'olfactory-conditioning-lab-v1/trial_plan.py': '6fed3f494d075339a7e5b001d43b2598640b189e038952fbd4d41200bacac587',
|
|
}
|
|
AP_ANCHORS = {
|
|
'protocol/preregistered.json': '02a03b536a9f14196ad4b92fa9565b592732c6c863e72eeaee89b3b0ea8ed37d',
|
|
'manifest.json': 'a40e433541a6c02f7ad95622e96646da9a90a93e4e43cb3abfde3846b2554f00',
|
|
'audit/final-status.json': '26de8371867d55253358a4be698de896445712e49e6eee8d99ce852fd0bf43ab',
|
|
'cases/D1--reference--epoch-001/spec.json': 'f0b8eddd78a2dc0e1d6fcae07489dcb24aa3c5bb14326367fa32cfdf960739ba',
|
|
}
|
|
PARENT_PINS = {
|
|
'D1': 'df328eddff4c735e08fae377bbd7c4cd0df288de60c6ebd34965e28c369fc084',
|
|
'D0': '6511e794c4f0f9929e81bd7363ad2314f0e44d9f95882883ab3644cf2b1a6fd9',
|
|
'frozen': 'ebebf86480d6fe9facb4f08689783a1b78143f4794fa11694d68dd9efb806373',
|
|
}
|
|
STALE_INPUT_KEYS = frozenset(('CS_window', 'cs_input_seed', 'cs_input_seed_bits', 'expected_CS_requests',
|
|
'mode', 'odor_id', 'odor_name', 'rate_sum_hz', 'source_panel_row', 'source_panel_split',
|
|
'target_count', 'seed_scope', 'outer_arm', 'outer_epoch_id'))
|
|
CASE_CHANGED_KEYS = frozenset(('name', 'steps', 'max_steps', 'duration_ms', 'budget_root',
|
|
'output_budget_bytes', 'outer_protocol', 'within_trial_weight_updates'))
|
|
CASE_ADDED_KEYS = frozenset(('boundary_diagnostic', 'input_policy', 'scientific_learning_claimed',
|
|
'diagnostic_boundary_tick', 'evaluation_stream_id', 'cue', 'probe_fork_discard'))
|
|
|
|
|
|
def _require(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _canonical(value):
|
|
return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False)
|
|
|
|
|
|
def _sha(value):
|
|
return hashlib.sha256(_canonical(value).encode('utf-8')).hexdigest()
|
|
|
|
|
|
def evaluation_seeds():
|
|
"""Exact frozen seed/stream rules; derive only, never draw random numbers."""
|
|
result = {}
|
|
for cue, odor in (('A', '4-ethylphenol'), ('B', '3-hexanone')):
|
|
key = [BLOCK, 'evaluation', 'CS', cue, '00', 'base-probe']
|
|
digest = _sha(['phase4an-seed-registry-v1', MASTER, 'evaluation', key])
|
|
result[cue] = dict(cue=cue, odor=odor, master=MASTER, split='evaluation', key=key,
|
|
seed=int(digest[:16], 16) & ((1 << 63)-1), seed_bits=63, derivation_sha256=digest,
|
|
stream_id='evaluation:' + _sha(['phase4an-trial-plan-v1-draft', 'stream-id', MASTER, 'evaluation', key]))
|
|
return result
|
|
|
|
|
|
def build_plan():
|
|
"""Fixed proposal; all future hashes and identity checks remain external."""
|
|
calls = []
|
|
|
|
def add(name, runner, mode, steps, parent, offset, overlay, policy='empty_idle', **extra):
|
|
group = name
|
|
if '--updated--' in name or '--old-C--' in name:
|
|
arm = name.split('--')[0]
|
|
group = ('mechanical' if arm == 'mechanical' else 'AP-'+arm)+'-suffix'
|
|
calls.append(dict(call_id=name, runner=runner, diagnostic_mode=mode, steps=steps,
|
|
case_group=group,
|
|
parent=parent, global_tick_offset=offset, global_next_tick=offset+steps,
|
|
overlay_after_role=overlay, input_policy=policy, scientific_learning_claimed=False, **extra))
|
|
|
|
imports = {arm: dict(role=f'AP.{arm}.reference.epoch001',
|
|
checkpoint_manifest=AP_ROOT+f'/runs/{arm}--reference--epoch-001/checkpoint-manifest.json',
|
|
sha256=pin, steps_already_executed=1200) for arm, pin in PARENT_PINS.items()}
|
|
for arm in ('D1', 'D0', 'frozen'):
|
|
parent = imports[arm]['role']
|
|
after = parent+'.original_E.after'
|
|
for mode in ('reference', 'light'):
|
|
add(f'{arm}--updated--{mode}', 'updated_continuation', mode, 1000, parent, 1200, after,
|
|
transition='epoch_update', overlay_before_role=parent+'.recorded_overlay')
|
|
if arm == 'D1':
|
|
add('D1--live-oracle', 'boundary_oracle', 'reference', 2200, None, 0, after,
|
|
policy='replay_AP_prefix_then_empty_idle', prefix_replay_parent=parent,
|
|
overlay_before_role=parent+'.recorded_overlay', diagnostic_boundary_tick=1200,
|
|
within_trial_weight_updates=True, restore_used=False)
|
|
else:
|
|
add(f'{arm}--old-C--reference', 'runtime_continuation', 'reference', 1000, parent,
|
|
1200, parent+'.recorded_overlay', transition='same_weight')
|
|
add('performance--natural-light-fresh', 'runtime_light', 'light', 1200, None, 0,
|
|
'AP.D1.reference.epoch001.recorded_overlay', policy='replay_AP_D1_epoch001_input',
|
|
input_replay_source='AP.D1.reference.epoch001', performance_training_cycle=True,
|
|
regression_against='AP.D1.reference.epoch001', new_learning=False, input_sampled=False)
|
|
add('performance--evaluation-A-light-fresh', 'fresh_probe', 'light', 1200, None, 0,
|
|
'AP.initial_overlay', policy='externally_pinned_CS_probe', cue='A',
|
|
performance_probe_cycle=True, probe_fork_discard=True, neural_reset=True,
|
|
US=False, plastic_update=False, feedback_to_idle=False,
|
|
evaluation_stream_id=evaluation_seeds()['A']['stream_id'])
|
|
add('mechanical--fresh-parent', 'runtime_light', 'reference', 1200, None, 0,
|
|
'AP.initial_overlay', policy='synthetic_one_KC_plus_four_PAM15_at_tick1199',
|
|
inherited_learned_delta=False, olfactory_input=False)
|
|
for mode in ('reference', 'light'):
|
|
add(f'mechanical--updated--{mode}', 'updated_continuation', mode, 1000,
|
|
'mechanical--fresh-parent', 1200, 'mechanical.original_E.after', transition='epoch_update',
|
|
overlay_before_role='AP.initial_overlay')
|
|
add('mechanical--live-oracle', 'boundary_oracle', 'reference', 2200, None, 0,
|
|
'mechanical.original_E.after', policy='replay_mechanical_prefix_then_empty_idle',
|
|
prefix_replay_parent='mechanical--fresh-parent', overlay_before_role='AP.initial_overlay',
|
|
diagnostic_boundary_tick=1200, within_trial_weight_updates=True, restore_used=False)
|
|
parent = imports['D1']['role']
|
|
after = parent+'.original_E.after'
|
|
add('D1--idle--1s', 'updated_continuation', 'light', 10000, parent, 1200, after,
|
|
transition='epoch_update', overlay_before_role=parent+'.recorded_overlay')
|
|
add('D1--idle--9s', 'updated_continuation', 'light', 90000, 'D1--idle--1s', 11200, after,
|
|
transition='idle', overlay_before_role=after)
|
|
for label, branch, offset, transition in (('t0', parent, 1200, 'epoch_update'),
|
|
('t1', 'D1--idle--1s', 11200, 'idle'), ('t10', 'D1--idle--9s', 101200, 'idle')):
|
|
for cue in ('A', 'B'):
|
|
add(f'D1--probe--{label}--{cue}', 'updated_continuation', 'light', 1200, branch, offset,
|
|
after, policy='externally_pinned_CS_probe', transition=transition, cue=cue,
|
|
overlay_before_role=parent+'.recorded_overlay' if label == 't0' else after,
|
|
probe_fork_discard=True, feedback_to_idle=False, neural_reset=False, US=False,
|
|
plastic_update=False, evaluation_stream_id=evaluation_seeds()[cue]['stream_id'])
|
|
zero_root = '/data/digital-drosophila/experiments/'
|
|
return dict(schema=SCHEMA, status='DRAFT-NOT-REGISTERED', calls=calls, imported_parents=imports,
|
|
initial_overlay=dict(role='AP.initial_overlay', registration_field='initial_overlay',
|
|
overlay_file=dict(path=zero_root+'phase4am-wholebrain-plastic-interface/overlays/zero.npz',
|
|
sha256='0b219eab09c30dc3af31fef7e6395a9bec1234090abbb72db096da833fed82c5'),
|
|
overlay_manifest=dict(path=zero_root+'phase4an-engineering-validation/overlays/zero.json',
|
|
sha256='11b668978f3070dad26d99d9310bebefa0731fa0d4e361e8dc0ef02a213277c2')),
|
|
source_sha256=deepcopy(SOURCE_PINS), AP_anchors_sha256=deepcopy(AP_ANCHORS),
|
|
AP_source_commit='0bcce983324ecfe1c75826a23e94ff932ad5e883', evaluation_streams=evaluation_seeds(),
|
|
mechanical_update=dict(coordinator='original_E', update='original_U', parent='mechanical--fresh-parent',
|
|
before_overlay_role='AP.initial_overlay', steps=1200, us_tick=1199, d_command=1.0,
|
|
frozen=False, compartment=COMPARTMENT, trace_backend='torch'),
|
|
performance_update=dict(coordinator='original_E', update='original_U', parent='performance--natural-light-fresh',
|
|
before_overlay_role='AP.D1.reference.epoch001.recorded_overlay', steps=1200, us_tick=900,
|
|
d_command=1.0, frozen=False, compartment=COMPARTMENT, trace_backend='torch',
|
|
required_regression=['all_events', 'final_state', 'delta_after'],
|
|
timing_scope='before_brain_spawn_through_original_E_exit_and_independent_CPU_validation'),
|
|
cpu_children=[*[dict(role='verify_AP_original_E', parent=imports[a]['role']) for a in imports],
|
|
dict(role='performance_original_E'), dict(role='verify_performance_original_E'),
|
|
dict(role='mechanical_original_E'), dict(role='verify_mechanical_original_E')],
|
|
sampler_child=dict(role='evaluation_sampler', execution='GPU_private_generator',
|
|
uniform_batch_calls=2, cues=['A', 'B']),
|
|
counts=dict(new_brain_calls=23, new_brain_ticks=125200, distinct_case_groups=17, imported_brain_calls=3,
|
|
imported_brain_ticks=3600, covered_brain_ticks=128800, cpu_children=7, gpu_sampler_children=1, total_children=31,
|
|
new_CS_uniform_batch_calls=2, raw_uniform_draw_count=None, raw_draw_count_known=False),
|
|
budget_proposal=dict(output_bytes=4*2**30, free_reserve_bytes=10*2**30,
|
|
child_timeout_seconds=1800, workflow_timeout_seconds=7200, approved=False,
|
|
on_limit='stop_and_report_incomplete; never_shorten_trial_or_change_gate'),
|
|
protected_contract=dict(brain_model='unchanged', APL='unchanged', delay='unchanged',
|
|
sign='unchanged', threshold='unchanged', eta='unchanged', candidate_mask_count=14232),
|
|
external_requirements=['authenticate_all_source_protocol_checkpoint_and_overlay_bytes',
|
|
'authenticate_candidate_endpoints_KC_membership_and_exact_PAM15_membership',
|
|
'bind_original_E_outputs_and_CPU_recomputation_receipts',
|
|
'register_new_sources_cases_inputs_budget_and_exact_execution_authority',
|
|
'sample_two_evaluation_CS_batches_with_pinned_real_generator_then_reuse_at_three_times'],
|
|
brain_executed=False, inputs_sampled=False, acquisition_executed=False,
|
|
full_retention_experiment=False, scientific_learning_claimed=False,
|
|
external_identity_verified_here=False, execution_authorized=False)
|
|
|
|
|
|
def validate_plan(plan):
|
|
_require(_canonical(plan) == _canonical(build_plan()), 'Plan differs from fixed independent reconstruction')
|
|
return True
|
|
|
|
|
|
def derive_case_spec(template, call_id, *, budget_root):
|
|
"""Narrow derivation of the actual AP D1 template; no claimed file checking."""
|
|
_require(_sha(template) == TEMPLATE_PAYLOAD_SHA256, 'Not the pinned AP template payload')
|
|
_require(type(budget_root) is str, 'Output root must be an absolute POSIX string')
|
|
root = PurePosixPath(budget_root)
|
|
_require(root.is_absolute() and '..' not in root.parts and
|
|
str(root) not in ('/', AP_ROOT) and PurePosixPath(AP_ROOT) not in root.parents and
|
|
root not in PurePosixPath(AP_ROOT).parents, 'Use a distinct absolute non-ancestor output root')
|
|
call = next((c for c in build_plan()['calls'] if c['call_id'] == call_id), None)
|
|
_require(call is not None, 'Unknown fixed call')
|
|
result = {k: deepcopy(v) for k, v in template.items() if k not in STALE_INPUT_KEYS}
|
|
result.update(name=call['case_group'], steps=call['steps'], max_steps=call['steps'], duration_ms=call['steps']*.1,
|
|
budget_root=str(root), output_budget_bytes=4*2**30, outer_protocol=SCHEMA,
|
|
within_trial_weight_updates=call.get('within_trial_weight_updates', False),
|
|
boundary_diagnostic=call['case_group'], input_policy=call['input_policy'], scientific_learning_claimed=False)
|
|
if 'diagnostic_boundary_tick' in call:
|
|
result['diagnostic_boundary_tick'] = call['diagnostic_boundary_tick']
|
|
if 'cue' in call:
|
|
stream = evaluation_seeds()[call['cue']]
|
|
result.update(mode='physiological', cue=call['cue'], odor_name=stream['odor'], CS_window=[0, 1000],
|
|
cs_input_seed=stream['seed'], cs_input_seed_bits=63, evaluation_stream_id=stream['stream_id'],
|
|
probe_fork_discard=True, seed_scope='reset seed unchanged; full63 evaluation CS seed used by external sampler')
|
|
return result
|
|
|
|
|
|
def mechanical_requests(candidate_map, pam15_ids, *, identity_binding):
|
|
"""Caller authenticates mapping/DAN membership; only array schema checked here.
|
|
|
|
Select minimum canonical KC index, never based on outcome. Returned requests
|
|
are synthetic deterministic force requests, not odor samples or accepted spikes.
|
|
"""
|
|
roles = {'edge_id', 'source_neuron_idx', 'target_neuron_idx', 'target_compartment_verbatim'}
|
|
_require(type(candidate_map) is dict and set(candidate_map) == roles, 'Exact candidate map schema required')
|
|
for key, array in candidate_map.items():
|
|
_require(isinstance(array, np.ndarray) and array.shape == (CANDIDATES,), 'Full candidate arrays required')
|
|
expected = np.int64 if key == 'edge_id' else np.int32
|
|
_require(array.dtype.kind == 'U' if key == 'target_compartment_verbatim' else array.dtype == expected,
|
|
'Candidate dtype changed')
|
|
edge = candidate_map['edge_id']
|
|
_require(np.all(edge >= 0) and np.all(edge < 25582938) and np.all(edge[1:] > edge[:-1]), 'Canonical edge order required')
|
|
for key in ('source_neuron_idx', 'target_neuron_idx'):
|
|
_require(np.all((candidate_map[key] >= 0) & (candidate_map[key] < N)), 'Neuron index out of range')
|
|
_require(isinstance(pam15_ids, np.ndarray) and pam15_ids.dtype == np.int32 and pam15_ids.shape == (4,)
|
|
and np.all((pam15_ids >= 0) & (pam15_ids < N)) and np.all(pam15_ids[1:] > pam15_ids[:-1]),
|
|
'Four canonical sorted unique int32 PAM15 indices required')
|
|
_require(type(identity_binding) is dict and set(identity_binding) == {'candidate_map_sha256', 'pam15_membership_sha256'}
|
|
and all(type(v) is str and len(v) == 64 and all(c in '0123456789abcdef' for c in v)
|
|
for v in identity_binding.values()), 'External identity digest descriptors required')
|
|
sources = candidate_map['source_neuron_idx'][candidate_map['target_compartment_verbatim'] == COMPARTMENT]
|
|
_require(len(sources) > 0, 'No exact y5B\'2a candidate rows')
|
|
selected = int(sources.min())
|
|
_require(selected not in pam15_ids, 'KC/DAN identity overlap')
|
|
node = np.sort(np.concatenate((np.array([selected], dtype=np.int32), pam15_ids)))
|
|
audit = dict(selected_KC_source_neuron_idx=selected, PAM15_neuron_idx=pam15_ids.tolist(), tick=1199,
|
|
request_count=5, compartment=COMPARTMENT, identity_binding=deepcopy(identity_binding),
|
|
selection='minimum canonical source_neuron_idx among exact compartment candidate rows',
|
|
input_arrays={k: dict(dtype=a.dtype.str, shape=list(a.shape), raw_c_order_sha256=
|
|
hashlib.sha256(a.tobytes(order='C')).hexdigest()) for k, a in candidate_map.items()},
|
|
external_identity_verified_here=False, olfactory_input=False, sampled=False, accepted_spikes_verified=False,
|
|
before_overlay_role='AP.initial_overlay', inherited_learned_delta=False, scientific_learning_claimed=False)
|
|
return dict(tick=np.full(5, 1199, dtype=np.int32), neuron_idx=node), audit
|