diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/CONDITIONING_INTERFACE_PROTOCOL.md b/models/male-cns-lif/olfactory-conditioning-lab-v1/CONDITIONING_INTERFACE_PROTOCOL.md new file mode 100644 index 0000000..ee7f68d --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/CONDITIONING_INTERFACE_PROTOCOL.md @@ -0,0 +1,70 @@ +# AO — bounded two-epoch conditioning-interface validation + +This independent branch does not replace the frozen reference, runtime or any +Phase 1–AN result. It tests the installation interface for the existing frozen +eligibility-v2 rule. It is not a new dynamics hypothesis or a learned model. + +## Fixed, pre-outcome selection + +- Master seed: `20260930`; no seed search or outcome-based retries. +- First canonical block: physiological / PAM15 / reinforce_A / rep01. +- First normal paired epoch requesting a US, and its immediate successor. +- Both selected epochs are odor A (4-ethylphenol), not an A/B discrimination + test. Exact recorded L glomerular/PN mapping and PN column order are retained. +- Each epoch is 120 ms: 100 ms CS plus 20 ms tail, US at tick 900. +- D1, D0 and frozen arms receive identical saved CS and DAN requests. The + scheduled artificial D scalar is not measured dopamine concentration and is + not multiplied by actual DAN request acceptance. Delivery is reported apart. +- Two executors (reference and light), two fresh epochs per arm: 12 calls, + 14,400 timesteps. This does not authorize the full conditioning inventory. + +## Immutable neural semantics and authority + +Each epoch resets V, g, refractory counters, APL state and both delay queues; +the zero-initialized eligibility trace is computed from actual sealed spikes. +Only the float32-mV plastic delta carries between epochs. No update occurs +inside an epoch. Original graph, base weights, transmitter signs, numerical +precision, APL parameters, frozen v2 rule and complete 14,232-edge mask remain +unchanged. Full graph is 166,700 neurons and 25,582,938 edges. + +Registration pins source snapshots, actual graph/APL/coefficient files, frozen +L static resources, H mappings, the complete generated semantic plan and the +12 selected call descriptors before CUDA sampling. Sampling receipts bind the +two generated request arrays before any brain call. Every input file is shared +byte-for-byte between arms and executors. The full 63-bit input seed is separate +from the inherited conservative 31-bit neural reset-control value. + +A fixed first-seed CUDA check executes only the four original L sampling AST +statements, compares sparse requests, and compares full63-bit versus low32-bit +uniform matrices. This is a sampler test, not a neural trial or seed search. +Two production and three diagnostic matrices are drawn; each is 1200×PN count, +at most 2,000,000 uniform values. No continuous-time Poisson claim is made. + +Each neural output is immediately sealed. Outside that seal, the coordinator +binds original inputs, checkpoint events, actual v2 recomputation and the new +delta snapshot. `parent_snapshot_sha256` means the previous **overlay NPZ file**, +not its manifest or the neural checkpoint. Epoch 2 must install that exact +snapshot; D0/frozen deltas must stay positive-zero bitwise. + +## Finite gates and reporting limits + +Required: all 12 processes complete; six reference/light pairs have identical +sparse events, counts, final V/g/ref and terminal queues/APL; deltas are exact; +D0/frozen identity holds; epoch 2 actually installs epoch 1 delta. D1 must have +nonzero actual update coverage or the result is `INCOMPLETE`, never rescued by +another input. Nonzero delta is interface coverage, not evidence of memory. + +The light executor does not record intermediate floating states; no all-tick +bitwise claim is inferred. This validation has only one training replicate, two +A epochs, no independent behavioral evaluation, and no acquisition/reversal/ +retention success claim. The post-update retention boundary is still unsupported. + +## Resource and failure policy + +Use only RTX4090 UUID GPU-79f3a85d-adce-7833-7003-6488140ca596; admit a child only +when all three ComfyUI queues are empty. Never stop or evict services. Cap new +output at 4 GiB plus 10 GiB free reserve and each child at 1800 seconds. Preserve +partial output and stop on any failure; no automatic brain retry. End by +rechecking AN and AM seals and the inherited 28-root integrity inventory. Five +old frozen reference results are retained from sealed AM evidence, not claimed +to have been rerun by this small interface gate. diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_epoch.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_epoch.py new file mode 100644 index 0000000..82c23ee --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_epoch.py @@ -0,0 +1,312 @@ +"""Exclusive postprocessing of one externally authorized sealed fresh epoch. + +No brain execution, source selection, input generation or weight installation. +The external caller must derive and pin authority from its registered row and +successful process receipt; a SHA supplied to this helper is not self-proving +registration authority. Production uses the actual pinned C/R checkpoint and +overlay readers. Small synthetic tests exercise only the private core. +""" +from __future__ import annotations + +import hashlib +import json +import math +from pathlib import Path +import sys +from types import ModuleType + +import numpy as np + + +SCHEMA = 'phase4an-epoch-postprocess-authority-v1' +SOURCE_PINS = { + 'runtime_light_source': '522b8c31ea938890fe15364fccd25ccc5bded54101754909f255532e7ad01aae', + 'runtime_continuation_source': '0b8e8e0a6a23a079dc62a88355ce355d53b3ae22f6665519a8f6aef2dcfc5aba', + 'conditioning_update_source': '96efff678cff910654eb08e86f1a17789e4e5589e340345f419c7027a7d547e2', + 'eligibility_bridge_source': '174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61', + 'phase4j_source': 'abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e', + 'am_overlay_source': '7f891d0ba79b393c96e75fd2dee2eddc51e57d8213eaac2779703bc2c18712b4', +} +ROLES = frozenset(('spec', 'requests', 'overlay_before', 'overlay_manifest_before', 'state', + 'kc', 'canonical_kc', 'candidate_map', 'mask', 'base_candidate', *SOURCE_PINS)) +GRAPH_FILES = ('nodes.npy', 'edge_id.npy', 'src.npy', 'dst.npy', 'signed_raw.npy', 'ptr.npy', 'weights.npy') +APL_FILES = ('input_matrix.npy', 'output_matrix.npy', 'kc_globals.npy', 'apl_globals.npy') +DIMENSIONS = (166700, 25582938, 4064, 14232) +_MODULES = {} + + +def _require(ok, message): + if not ok: + raise ValueError(message) + + +def _sha(value): + return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value) + + +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 _read(path): + def pairs(items): + value = {} + for key, item in items: + _require(key not in value, 'Duplicate JSON key: ' + key) + value[key] = item + return value + return json.loads(Path(path).read_text(encoding='utf-8'), object_pairs_hook=pairs, + parse_constant=lambda value: (_ for _ in ()).throw(ValueError('Nonfinite JSON: ' + value))) + + +def _save(path, value): + with Path(path).open('x', encoding='utf-8') as stream: + json.dump(value, stream, sort_keys=True, indent=2, allow_nan=False) + stream.write('\n') + + +def _path(value): + _require(type(value) is str and Path(value).is_absolute(), 'Input paths must be explicit and absolute') + path = Path(value) + _require(path.is_file() and not path.is_symlink(), 'Input must be an existing regular, unlinked file') + return path.resolve(strict=True) + + +def _check_files(pins): + for path, expected in pins.items(): + _require(_sha(expected) and _digest(path) == expected, 'Input/source SHA mismatch: ' + str(path)) + + +def _authority(path, expected): + path = _path(str(path)) + _require(_sha(expected) and _digest(path) == expected, 'Authority differs from external SHA') + value = _read(path) + keys = {'schema', 'epoch_id', 'diagnostic_mode', 'command', 'checkpoint_dir', + 'checkpoint_manifest_sha256', 'files', 'frozen_inputs'} + _require(type(value) is dict and set(value) == keys and value['schema'] == SCHEMA, 'Invalid authority fields/schema') + _require(type(value['epoch_id']) is str and value['epoch_id'] and value['epoch_id'].strip() == value['epoch_id'], 'Invalid epoch identity') + _require(value['diagnostic_mode'] in ('light', 'reference'), 'Invalid diagnostic mode') + command = value['command'] + _require(type(command) is dict and set(command) == {'steps', 'compartment', 'us_tick', 'd_command', 'frozen', 'trace_backend'}, 'Invalid command fields') + _require(type(command['steps']) is int and command['steps'] > 0, 'Invalid command steps') + _require(command['us_tick'] is None or (type(command['us_tick']) is int and 0 <= command['us_tick'] < command['steps']), 'Invalid US tick') + _require(type(command['d_command']) in (int, float) and math.isfinite(command['d_command']) and command['d_command'] in (0, 1), 'Invalid scheduled D') + _require(command['us_tick'] is not None or command['d_command'] == 0, 'No US requires D=0') + _require(type(command['frozen']) is bool and command['trace_backend'] in ('numpy', 'torch'), 'Explicit frozen/backend required') + _require(command['compartment'] in ("B2B'2a", 'a1', "y2a'1", "y5B'2a"), 'Explicit frozen compartment required') + _require(type(value['files']) is dict and set(value['files']) == ROLES, 'Missing/unknown authority file roles') + paths, pins = {}, {path: expected} + for role, entry in value['files'].items(): + _require(type(entry) is dict and set(entry) == {'path', 'sha256'} and _sha(entry['sha256']), 'Invalid file authority: ' + role) + location = _path(entry['path']) + _require(location not in pins or pins[location] == entry['sha256'], 'Conflicting hashes for the same file') + paths[role], pins[location] = location, entry['sha256'] + _require(type(value['frozen_inputs']) is dict and value['frozen_inputs'], 'Full frozen input pins are required') + for name, sha in value['frozen_inputs'].items(): + location = _path(name) + _require(_sha(sha) and (location not in pins or pins[location] == sha), 'Invalid/conflicting frozen input pin') + pins[location] = sha + checkpoint = Path(value['checkpoint_dir']) + _require(checkpoint.is_absolute() and checkpoint.is_dir() and not checkpoint.is_symlink(), 'Explicit existing checkpoint directory required') + checkpoint = checkpoint.resolve(strict=True) + manifest = checkpoint / 'checkpoint-manifest.json' + _require(_sha(value['checkpoint_manifest_sha256']), 'External checkpoint manifest SHA required') + _require(manifest not in pins or pins[manifest] == value['checkpoint_manifest_sha256'], 'Conflicting checkpoint manifest pins') + pins[manifest] = value['checkpoint_manifest_sha256'] + _require(paths['state'] == checkpoint / 'state.npz', 'Events must come from this checkpoint state.npz') + _check_files(pins) + return value, paths, pins, checkpoint, path + + +def _load_source(path, expected, role): + raw = path.read_bytes() + _require(hashlib.sha256(raw).hexdigest() == expected, 'Pinned source changed: ' + role) + name = '_conditioning_epoch_' + role + if name in _MODULES: + saved_path, saved_sha, module = _MODULES[name] + _require(saved_path == path and saved_sha == expected and sys.modules.get(name) is module, + 'Source module identity collision: ' + role) + return module + _require(name not in sys.modules, 'Unverified source module collision: ' + role) + module = ModuleType(name) + module.__file__ = str(path) + sys.modules[name] = module + try: + exec(compile(raw, str(path), 'exec'), module.__dict__) + except BaseException: + sys.modules.pop(name, None) + raise + _MODULES[name] = (path, expected, module) + return module + + +def _production_dependencies(paths, pins): + for role, expected in SOURCE_PINS.items(): + _require(pins[paths[role]] == expected, 'Authority cannot replace frozen source: ' + role) + _require(paths['runtime_continuation_source'].parent / 'runtime_light.py' == paths['runtime_light_source'], + 'Continuation must load its declared adjacent frozen runtime') + c = _load_source(paths['runtime_continuation_source'], SOURCE_PINS['runtime_continuation_source'], 'continuation') + _require(Path(c.R.__file__).resolve() == paths['runtime_light_source'] and + _digest(c.R.__file__) == SOURCE_PINS['runtime_light_source'], 'Loaded actual runtime differs from authority') + u = _load_source(paths['conditioning_update_source'], SOURCE_PINS['conditioning_update_source'], 'update') + return c, c.R, u + + +def _npz(path, keys): + with np.load(path, allow_pickle=False) as archive: + _require(len(archive.files) == len(keys) and set(archive.files) == set(keys), 'Unexpected NPZ schema: ' + str(path)) + return {key: archive[key].copy() for key in keys} + + +def _same(a, b): + return a.dtype == b.dtype and a.shape == b.shape and a.tobytes() == b.tobytes() + + +def _vector(a, dtype, count, label): + _require(a.dtype == np.dtype(dtype) and a.shape == (count,), 'Invalid dtype/shape: ' + label) + + +def _coordinate(value, paths, pins, checkpoint, authority_path, out_dir, dependencies, dimensions): + """Private core: explicit tiny-fixture dependencies only in synthetic tests.""" + c, r, u = dependencies + n, edges, kc_count, candidates = dimensions + out = Path(out_dir).resolve(strict=False) + protected = {checkpoint, *(path.parent for path in pins)} + _require(not out.exists() and not any(out.is_relative_to(root) for root in protected), + 'Output must be new and outside checkpoint/source/data/input parent directories') + _require(out.parent.is_dir(), 'Output parent must already exist') + source_sha = _digest(__file__) + spec = r.validate_spec(_read(paths['spec']), paths['spec'].parent) + command = value['command'] + _require(type(spec['steps']) is int and spec['steps'] == command['steps'], 'Command steps differ from authorized spec') + _require(paths['spec'].name == 'spec.json' and paths['requests'] == paths['spec'].parent / 'requests.npz', 'Actual case input filenames differ') + graph = Path(spec['graph_dir']).resolve(strict=True) + apl = Path(spec['apl_bundle_dir']).resolve(strict=True) + required_inputs = [graph / name for name in GRAPH_FILES] + [apl / name for name in APL_FILES] + [Path(spec['coefficients_path']).resolve(strict=True)] + _require(all(path in pins for path in required_inputs), 'Full graph/APL/coefficient authority is incomplete') + _require(paths['canonical_kc'] == apl / 'kc_globals.npy', 'Canonical KC must be the declared frozen APL/KC membership') + old_meta = _read(paths['overlay_manifest_before']) + for role, prefix in (('mask', 'mask'), ('base_candidate', 'base_candidate'), ('am_overlay_source', 'am_overlay')): + _require(Path(old_meta[prefix + '_path']).resolve(strict=True) == paths[role] and + old_meta[prefix + '_sha256'] == pins[paths[role]], 'Overlay authority differs: ' + role) + aj = Path(old_meta['aj_model_dir']).resolve(strict=True) + for relative, expected in r.PINS.items(): + path = (aj.parent / relative).resolve(strict=True) + _require(pins.get(path) == expected, 'Missing exact frozen dependency pin: ' + relative) + record = c.load_checkpoint(checkpoint, value['checkpoint_manifest_sha256']) + _require(record['fresh'] is True and record['global_tick_offset'] == 0 and record['steps'] == command['steps'], 'Only complete fresh epochs may produce an update') + summary, provenance = record['summary'], record['provenance'] + _require(summary['nodes'] == n and summary['edges'] == edges and summary['diagnostic_mode'] == value['diagnostic_mode'], 'Runtime population/mode differs from authority') + _require(record['manifest']['files_sha256']['state.npz'] == pins[paths['state']], 'State/events are not bound by the expected checkpoint seal') + for role in ('spec', 'requests', 'overlay_before', 'overlay_manifest_before'): + _require(provenance['input_sha256'].get(str(paths[role])) == pins[paths[role]], 'Runtime did not bind authorized ' + role) + _require(_digest(checkpoint / 'overlay-input.npz') == pins[paths['overlay_before']] and + _digest(checkpoint / 'overlay-manifest.json') == pins[paths['overlay_manifest_before']], 'Sealed runtime used a different overlay or manifest') + helper, before, checked_meta, _ = r.load_overlay(paths['overlay_before'], paths['overlay_manifest_before']) + _require(checked_meta == old_meta, 'Overlay manifest changed during validation') + mapping = _npz(paths['candidate_map'], ('edge_id', 'source_neuron_idx', 'target_neuron_idx', 'target_compartment_verbatim')) + mask = np.load(paths['mask'], allow_pickle=False) + kc = np.load(paths['kc'], allow_pickle=False) + canonical_kc = np.load(paths['canonical_kc'], allow_pickle=False) + base = np.load(paths['base_candidate'], allow_pickle=False) + _vector(mask, 'int64', candidates, 'complete mask') + _vector(kc, 'int32', kc_count, 'KC') + _vector(canonical_kc, 'int32', kc_count, 'canonical KC') + _vector(base, 'float32', candidates, 'original base mV') + _require(_same(kc, canonical_kc), 'Supplied KC differs from canonical membership bytes') + _require(_same(mapping['edge_id'], mask) and _same(before['edge_id'], mask), 'Candidate mapping/overlay differ from exact mask') + _require(_same(before['base_candidate_mv'], base), 'Overlay used different original base bytes') + for key in ('source_neuron_idx', 'target_neuron_idx'): + _vector(mapping[key], 'int32', candidates, key) + comps = mapping['target_compartment_verbatim'] + _require(comps.shape == (candidates,) and comps.dtype.kind == 'U', 'Compartment mapping must be an exact Unicode vector') + maps = {} + try: + for filename, dtype, length in (('nodes.npy', 'int32', n), ('edge_id.npy', 'int64', edges), + ('src.npy', 'int32', edges), ('dst.npy', 'int32', edges)): + array = np.load(graph / filename, allow_pickle=False, mmap_mode='r') + maps[filename] = array + _vector(array, dtype, length, 'full graph ' + filename) + _require(np.array_equal(maps['nodes.npy'], np.arange(n, dtype=np.int32)), 'Global nodes are not canonical') + _require(np.all((mask >= 0) & (mask < edges)) and np.all(mask[1:] > mask[:-1]), 'Invalid canonical edge mask') + _require(_same(maps['edge_id.npy'][mask], mask), 'Mask is not indexed by original global edge ID') + _require(_same(mapping['source_neuron_idx'], maps['src.npy'][mask]) and + _same(mapping['target_neuron_idx'], maps['dst.npy'][mask]), 'Candidate endpoints differ from actual graph') + finally: + for array in maps.values(): + array._mmap.close() + state = _npz(paths['state'], ('step', 'neuron_idx', 'counts', 'v', 'g', 'ref')) + _require(all(_same(state[key], record['state'][key]) for key in state), 'Event/state arrays differ from actual sealed reader') + _require(record['overlay']['delta_raw_sha256'] == r._raw(before['plastic_delta_mv']) and + record['overlay']['effective_weights_raw_sha256'] == summary['effective_weights_raw_sha256'], 'Runtime weight installation is not bound to delta_before') + args = dict(step=state['step'], neuron_idx=state['neuron_idx'], kc=kc, + candidate_edge_id=mask, candidate_sources=mapping['source_neuron_idx'], + candidate_targets=mapping['target_neuron_idx'], candidate_compartments=comps, + delta_before=before['plastic_delta_mv'], base_candidate=base, steps=command['steps'], n_neurons=n, + compartment=command['compartment'], us_tick=command['us_tick'], d_command=command['d_command'], + frozen=command['frozen'], bridge_source=paths['eligibility_bridge_source'], + phase4j_source=paths['phase4j_source'], trace_backend=command['trace_backend']) + result = u.update_epoch(**args) + _check_files(pins) + # The directory creation is the exclusive claim. A later failure remains a + # partial directory and can never silently be retried or overwritten. + out.mkdir() + try: + with (out / 'delta-after.npz').open('xb') as stream: + np.savez_compressed(stream, edge_id=mask, plastic_delta_mv=result.delta_after) + metadata = {key: old_meta[key] for key in ('schema', 'mask_path', 'mask_sha256', + 'base_candidate_path', 'base_candidate_sha256', 'aj_model_dir', 'am_overlay_path', 'am_overlay_sha256')} + metadata.update(overlay_name='learned', overlay_sha256=_digest(out / 'delta-after.npz'), + parent_snapshot_sha256=pins[paths['overlay_before']], eligibility_v2_source_sha256=SOURCE_PINS['phase4j_source']) + _save(out / 'overlay-manifest.json', metadata) + _save(out / 'update-receipt.json', result.receipt) + _, after, _, _ = r.load_overlay(out / 'delta-after.npz', out / 'overlay-manifest.json') + _require(_same(after['plastic_delta_mv'], result.delta_after) and _same(after['edge_id'], mask), 'Written learned snapshot failed actual runtime loader') + # Re-read the sealed neural state after writing, without altering its + # seal or appending coordinator output inside the runtime directory. + c.load_checkpoint(checkpoint, value['checkpoint_manifest_sha256']) + _check_files(pins) + _require(_digest(__file__) == source_sha, 'Coordinator source changed') + output_hashes = {name: _digest(out / name) for name in ('delta-after.npz', 'overlay-manifest.json', 'update-receipt.json')} + chain = dict(schema='phase4an-epoch-postprocess-chain-v1', complete=True, + epoch_id=value['epoch_id'], authority_file=str(authority_path), authority_sha256=pins[authority_path], + coordinator_source_sha256=source_sha, command=dict(command), diagnostic_mode=value['diagnostic_mode'], + before_overlay_file_sha256=pins[paths['overlay_before']], + before_overlay_manifest_sha256=pins[paths['overlay_manifest_before']], + after_overlay_file_sha256=output_hashes['delta-after.npz'], + after_overlay_manifest_sha256=output_hashes['overlay-manifest.json'], + parent_snapshot_sha256_semantics='delta-before overlay NPZ file SHA256, not neural checkpoint or manifest SHA', + source_neural_checkpoint_dir=str(checkpoint), + source_neural_checkpoint_manifest_sha256=value['checkpoint_manifest_sha256'], + source_state_file_sha256=pins[paths['state']], source_spec_sha256=pins[paths['spec']], + source_requests_sha256=pins[paths['requests']], + runtime_delta_before_raw_sha256=record['overlay']['delta_raw_sha256'], + runtime_effective_weights_raw_sha256=summary['effective_weights_raw_sha256'], + input_files_sha256={str(path): sha for path, sha in sorted(pins.items(), key=lambda item: str(item[0]))}, + output_files_sha256=output_hashes, + update_receipt_payload_sha256=result.receipt['payload_sha256'], + runtime_learned_manifest_loader_verified=True, input_and_checkpoint_bytes_reverified=True, + candidate_endpoints_equal_graph=True, KC_equals_supplied_canonical_membership=True, + authority_scope='caller must bind this externally pinned authority to its preregistered row, successful process, anatomical metadata and complete mask audit', + registration_authority_independently_proven=False, anatomical_identity_inferred_from_lengths=False, + neural_checkpoint_and_plastic_parent_are_distinct=True, + weights_installed=False, brain_executed=False, learning_pass_claimed=False, candidate_generated=False) + _save(out / 'chain-receipt.json', chain) + return chain + except BaseException as error: + if not (out / 'postprocess-error.json').exists(): + _save(out / 'postprocess-error.json', dict(complete=False, error_type=type(error).__name__, + message=str(error), partial_preserved=True, retry_forbidden=True, learning_pass_claimed=False)) + raise + + +def postprocess_epoch(authority_json, expected_authority_sha256, out_dir): + """Production entry: full 166700/25582938/4064/14232, no small-mode flag.""" + _require(sys.dont_write_bytecode, 'Use python -B; frozen sources must not receive bytecode writes') + value, paths, pins, checkpoint, authority_path = _authority(authority_json, expected_authority_sha256) + dependencies = _production_dependencies(paths, pins) + return _coordinate(value, paths, pins, checkpoint, authority_path, out_dir, dependencies, DIMENSIONS) diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_compare.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_compare.py new file mode 100644 index 0000000..7ca5874 --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_compare.py @@ -0,0 +1,247 @@ +"""Pure CPU comparisons for a separately authorized, finite input/update test. + +No files, checkpoint seals, source pins, parent trees or GPU are read here. +The coordinator must first verify subprocess receipts, externally pinned +checkpoint manifests, registered source/input files and learned-parent chains. +It must independently recompute update receipts with conditioning_update's +verify_receipt. These functions bind supplied arrays and declared metadata; +they cannot replace those external checks or prove functional learning. + +The frozen engineering comparator is reused unchanged. Reference diagnostics +must be complete. Light has events, final state and queues, not per-tick V/g +or APL histories; missing light observations are never fabricated as exact. +""" +from __future__ import annotations + +import hashlib +import json +import re + +import numpy as np + +import engineering_compare as C + + +N, E, N_KC, CANDIDATE_COUNT = 166700, 25582938, 4064, 14232 +COMPARATOR_SHA256 = "5a2403f11bc08d65a9d242ac2cf9686b850e0c3680d65659f785e9b6b50c0238" +PHASE4J_SHA256 = "abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e" +ELIGIBILITY_BRIDGE_SHA256 = "174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61" +MASK_SHA256 = "8b0fb6eb2e6c31da4d17dc20d890aa3f367e514f05642de5bad6b42a1611c05b" +AM_OVERLAY_SHA256 = "7f891d0ba79b393c96e75fd2dee2eddc51e57d8213eaac2779703bc2c18712b4" +MANIFEST_FIELDS = frozenset(("schema", "overlay_name", "overlay_sha256", "mask_path", "mask_sha256", + "base_candidate_path", "base_candidate_sha256", "aj_model_dir", "am_overlay_path", "am_overlay_sha256", + "parent_snapshot_sha256", "eligibility_v2_source_sha256")) + + +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 _sha(value): + return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest() + + +def _valid_sha(value): + return type(value) is str and re.fullmatch("[0-9a-f]{64}", value) is not None + + +def _evidence(value): + return dict(dtype=value.dtype.str, shape=list(value.shape), nbytes=int(value.nbytes), + raw_c_order_sha256=hashlib.sha256(value.tobytes(order="C")).hexdigest()) + + +def _integer(mapping, key, expected): + return type(mapping.get(key)) is int and mapping[key] == expected + + +def _request_input(value, steps): + _require(isinstance(value, np.ndarray) and value.dtype == np.int32 and value.ndim == 2 and value.shape[1] == 2, + "Expected input requests must be exact int32 [M,2] (local tick, global neuron)") + _require(np.all((value[:, 0] >= 0) & (value[:, 0] < steps)) and + np.all((value[:, 1] >= 0) & (value[:, 1] < N)), "Expected input request is out of range") + keys = value[:, 0].astype(np.int64) * N + value[:, 1] + _require(len(keys) < 2 or np.all(keys[1:] > keys[:-1]), "Expected requests must already be sorted unique") + + +def compare_epoch_pair(reference_record, light_record, *, expected_requests, expected_steps, + expected_effective_weight_sha256): + """Compare fresh reference/light records, not paths or self-asserted seals. + + Inputs are dictionaries returned by frozen engineering_compare.load_run. + The coordinator supplies the original registered request array and the + independently expected effective-weight raw SHA, not a value copied from + the summary being checked. Corrupt array schemas raise ValueError. Valid + but different records return numerical_exact/pass_all=False. pass_all is + ONLY the supplied-record identity check, not a learning or activity gate. + """ + _require(type(expected_steps) is int and 0 < expected_steps <= np.iinfo(np.int32).max, + "Explicit expected epoch steps required") + _require(_valid_sha(expected_effective_weight_sha256), "Expected effective-weight raw SHA is required") + _request_input(expected_requests, expected_steps) + _require(type(reference_record) is dict and type(light_record) is dict, + "Supply already authenticated loaded records, not paths") + C.validate_run(reference_record) + C.validate_run(light_record) + inherited = C.compare_fresh(light_record, reference_record) + checks = dict(inherited["checks"]) + for label, record, mode in (("reference", reference_record, "reference"), ("light", light_record, "light")): + summary = record.get("summary") + _require(type(summary) is dict, "Missing runtime summary") + checks[label + ".full_population"] = len(record["state"]["v"]) == N and _integer(summary, "nodes", N) and _integer(summary, "edges", E) + checks[label + ".expected_clock"] = (record["steps"] == expected_steps and record["global_tick_offset"] == 0 and + _integer(summary, "steps", expected_steps) and _integer(summary, "completed_ticks", expected_steps) and + _integer(summary, "explicit_steps_executed", expected_steps)) + checks[label + ".fresh_fixed_weight_scope"] = (summary.get("checkpoint_continuation_used") is False and + summary.get("within_trial_weight_updates") is False and summary.get("hidden_early_stop") is False and + summary.get("entire_neural_trial_executed") is True and summary.get("record_history") is False and + summary.get("parent_checkpoint_manifest_sha256") is None) + checks[label + ".mode_precision_delay"] = (summary.get("diagnostic_mode") == mode and + summary.get("precision") == "float64" and summary.get("apl_precision") == "float32" and + _integer(summary, "apl_delay_steps", 18) and summary.get("state_units") == "SI volts") + checks[label + ".complete_reported_finite"] = (summary.get("complete") is True and summary.get("pass_all") is True and + summary.get("all_finite") is True and _integer(summary, "first_failure_tick", -1)) + checks[label + ".effective_weight_authority"] = summary.get("effective_weights_raw_sha256") == expected_effective_weight_sha256 + checks[label + ".input_request_array"] = np.array_equal(record["requests"][:, :2], expected_requests.astype(np.int64)) + checks[label + ".request_and_spike_accounting"] = (summary.get("counts_match_sparse_events") is True and + _integer(summary, "total_spikes", len(record["state"]["step"])) and + _integer(summary, "requested", len(expected_requests)) and + _integer(summary, "accepted_requests", int(record["requests"][:, 3].sum(dtype=np.int64))) and + _integer(summary, "rejected_requests", int(record["requests"][:, 4].sum(dtype=np.int64)))) + checks[label + ".recorded_full_APL_terminal"] = (record.get("terminal_origin") == "recorded_terminal_queues" and + record["terminal"]["apl_pending"].shape == (18, N_KC)) + full = reference_record + full_summary = full["summary"] + checks["reference.full_diagnostics"] = (set(full["traces"]) == set(C.TRACE_FILES) and + full.get("state_hashes") is not None and full.get("queues") is not None and + all(full_summary.get(key) is True for key in + ("full_APL_history_recorded", "full_tick_state_hashes_recorded", "full_queue_hashes_recorded", "full_sequence_bitwise_audit_performed"))) + if full.get("queues") is not None: + checks["reference.queue_summary_accounting"] = ( + _integer(full_summary, "arriving_edges", sum(row["count"] for row in full["queues"])) and + _integer(full_summary, "accepted_arriving_edges", sum(row["accepted_count"] for row in full["queues"]))) + else: + checks["reference.queue_summary_accounting"] = False + light_summary = light_record["summary"] + checks["light.declared_recording_scope"] = (not light_record["traces"] and + light_record.get("state_hashes") is None and light_record.get("queues") is None and + all(light_summary.get(key) is False for key in + ("full_APL_history_recorded", "full_tick_state_hashes_recorded", "full_queue_hashes_recorded", "full_sequence_bitwise_audit_performed"))) + for label, summary in (("reference", full_summary), ("light", light_summary)): + inputs = summary.get("input_sha256") + checks[label + ".input_hash_schema"] = (type(inputs) is dict and + set(inputs) == {"base_weights.npy", "coefficients.json", "requests.npz", "spec.json"} and + all(_valid_sha(value) for value in inputs.values())) + for name in ("base_weights.npy", "coefficients.json"): + left = full_summary.get("input_sha256", {}).get(name) + right = light_summary.get("input_sha256", {}).get(name) + checks["pair.input." + name] = _valid_sha(left) and left == right + for name in ("arriving_edges", "accepted_arriving_edges"): + checks["pair.summary." + name] = (type(full_summary.get(name)) is int and + type(light_summary.get(name)) is int and full_summary[name] == light_summary[name]) + checks = {key: bool(value) for key, value in checks.items()} + exact = all(checks.values()) + return dict(schema="conditioning-preflight-epoch-comparison-v1", numerical_exact=exact, pass_all=exact, + status="EXACT-RECORDED-EPOCH-PAIR-NOT-LEARNING" if exact else "FAIL-EPOCH-COMPARISON", + checks=checks, failed_checks=[key for key, ok in checks.items() if not ok], + expected_steps=expected_steps, expected_requests=_evidence(expected_requests), + expected_effective_weight_raw_sha256=expected_effective_weight_sha256, + recorded_pair_availability=inherited["reference_comparisons_available"], + activity=inherited["activity"], nonvacuous_activity=inherited["nonvacuous_activity"], + activity_is_descriptive_not_a_new_gate=True, + full_reference_diagnostics_required=True, light_intermediate_states_observed=False, + light_per_tick_V_g_ref_identity_claimed=False, independent_finite_latch_replay_performed=False, + independent_CSR_queue_reconstruction_performed=False, + independent_wholebrain_CPU_oracle_performed=False, + expected_comparator_source_sha256=COMPARATOR_SHA256, source_pin_verified_here=False, + external_file_seals_process_and_input_authority_verified_here=False, + input_generator_or_parent_delta_chain_verified_here=False, + learning_pass_claimed=False, candidate_generated=False, + scope="Supplied fresh epoch records and original requests only; prerequisite external checkpoint/source/parent verification belongs to coordinator") + + +def _delta(value, name): + _require(isinstance(value, np.ndarray) and value.dtype == np.float32 and value.shape == (CANDIDATE_COUNT,) and + np.isfinite(value).all(), name + " must be the full finite float32 candidate delta vector") + + +def audit_delta_binding(delta_before, delta_after, installed_delta, *, update_receipt, + parent_snapshot_sha256, child_manifest): + """Bind already computed/installed arrays to receipt and learned metadata. + + parent_snapshot_sha256 means the actual PARENT OVERLAY NPZ file hash, + externally established by the coordinator, NOT a checkpoint manifest hash + and NOT a raw delta-array hash. installed_delta must come from the child + overlay subsequently checked by frozen runtime_light.load_overlay. + + This function does not read those files, authenticate anatomical IDs, + recompute eligibility-v2, establish clipping bounds, or prove a causal + behavioral effect. Call conditioning_update.verify_receipt separately + against actual sealed epoch events and complete pinned mappings. + """ + for name, value in (("delta_before", delta_before), ("delta_after", delta_after), ("installed_delta", installed_delta)): + _delta(value, name) + _require(_valid_sha(parent_snapshot_sha256), "Externally established parent overlay FILE SHA is required") + _require(type(update_receipt) is dict and type(child_manifest) is dict, "Receipt and manifest must be dictionaries") + _require(type(update_receipt.get("input_arrays")) is dict and type(update_receipt.get("outputs")) is dict and + type(update_receipt.get("source_pins")) is dict and type(update_receipt.get("command")) is dict and + type(update_receipt.get("semantics")) is dict and type(update_receipt.get("counts")) is dict, + "Incomplete epoch update receipt schema") + before, after, installed = (_evidence(value) for value in (delta_before, delta_after, installed_delta)) + command = update_receipt["command"] + counts = update_receipt["counts"] + checks = { + "receipt.self_hash": update_receipt.get("payload_sha256") == _sha({k: v for k, v in update_receipt.items() if k != "payload_sha256"}), + "receipt.schema": update_receipt.get("schema") == "phase4an-conditioning-update-draft-v1" and + update_receipt.get("status") == "DRAFT-COMPUTED-EPOCH-UPDATE-NOT-INSTALLED", + "receipt.original_source_pins": update_receipt["source_pins"] == { + "eligibility_bridge_sha256": ELIGIBILITY_BRIDGE_SHA256, "phase4j_sha256": PHASE4J_SHA256}, + "receipt.delta_before": _canonical(update_receipt["input_arrays"].get("delta_before")) == _canonical(before), + "receipt.delta_after": _canonical(update_receipt["outputs"].get("delta_after")) == _canonical(after), + "computed_installed_delta_bitwise": C.bit_equal(delta_after, installed_delta), + "delta.nonpositive": bool(np.all(delta_before <= 0) and np.all(delta_after <= 0)), + "delta.no_potentiation": bool(np.all(delta_after <= delta_before)), + "receipt.candidate_count": _integer(counts, "candidates", CANDIDATE_COUNT), + "receipt.changed_bit_count": _integer(counts, "actual_bitwise_changed_edges", int(np.count_nonzero(delta_before.view(np.uint32) != delta_after.view(np.uint32)))), + "receipt.changed_value_count": _integer(counts, "actual_value_changed_edges", int(np.count_nonzero(delta_before != delta_after))), + "receipt.explicit_scheduled_D": type(command.get("D_scheduled")) is float and command["D_scheduled"] in (0.0, 1.0) and + type(command.get("frozen")) is bool, + "receipt.epoch_boundary_only": update_receipt["semantics"].get("application") == "epoch_boundary_after_all_recorded_ticks; affects subsequent epoch only" and + update_receipt["semantics"].get("delta_dtype") == "float32" and update_receipt["semantics"].get("delta_units") == "mV", + "receipt.not_a_DAN_measurement": update_receipt["semantics"].get("D_definition") == "artificial scheduled scalar" and + update_receipt["semantics"].get("actual_DAN_delivery_observed") is False and + update_receipt["semantics"].get("actual_DAN_delivery_used_to_gate") is False, + "manifest.exact_fields": set(child_manifest) == MANIFEST_FIELDS, + "manifest.learned_schema": child_manifest.get("schema") == "phase4an-overlay-v1" and child_manifest.get("overlay_name") == "learned", + "manifest.parent_overlay_file_sha": child_manifest.get("parent_snapshot_sha256") == parent_snapshot_sha256, + "manifest.frozen_rule_mask_helper": child_manifest.get("eligibility_v2_source_sha256") == PHASE4J_SHA256 and + child_manifest.get("mask_sha256") == MASK_SHA256 and child_manifest.get("am_overlay_sha256") == AM_OVERLAY_SHA256, + "manifest.declared_file_hashes": all(_valid_sha(child_manifest.get(key)) for key in ("overlay_sha256", "base_candidate_sha256")), + "manifest.declared_paths": all(type(child_manifest.get(key)) is str and bool(child_manifest[key]) + for key in ("mask_path", "base_candidate_path", "aj_model_dir", "am_overlay_path")), + } + no_update = command.get("frozen") is True or command.get("D_scheduled") == 0.0 + checks["zero_D_or_frozen_identity"] = not no_update or C.bit_equal(delta_before, delta_after) + for name in ("weights_installed", "base_modified", "brain_dynamics_executed", "scientific_learning_pass_claimed", "candidate_generated"): + checks["receipt.scope." + name] = update_receipt.get(name) is False + checks = {key: bool(value) for key, value in checks.items()} + exact = all(checks.values()) + return dict(schema="conditioning-preflight-delta-binding-v1", pass_all=exact, + status="EXACT-DELTA-ARRAY-BINDING-NOT-LEARNING" if exact else "FAIL-DELTA-BINDING", + checks=checks, failed_checks=[key for key, ok in checks.items() if not ok], + delta_before=before, computed_delta_after=after, installed_delta=installed, + parent_overlay_file_sha256=parent_snapshot_sha256, + declared_child_overlay_file_sha256=child_manifest.get("overlay_sha256"), + update_receipt_payload_sha256=update_receipt.get("payload_sha256"), + actual_changed_edges=int(np.count_nonzero(delta_before != delta_after)), + actual_bitwise_changed_edges=int(np.count_nonzero(delta_before.view(np.uint32) != delta_after.view(np.uint32))), + delta_nonzero_or_change_is_not_learning_evidence=True, + eligibility_v2_independently_recomputed_here=False, full_mask_identity_verified_here=False, + candidate_order_or_compartment_mapping_verified_here=False, floor_bound_independently_verified_here=False, + file_bytes_and_parent_snapshot_authority_verified_here=False, installation_in_runtime_verified_here=False, + actual_DAN_delivery_verified_here=False, learning_pass_claimed=False, candidate_generated=False, + scope="Supplied float32 mV arrays and declared parent/update metadata only; coordinator must verify real files, v2 recomputation and next-epoch installation") diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_inputs.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_inputs.py new file mode 100644 index 0000000..aa5a232 --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_inputs.py @@ -0,0 +1,288 @@ +"""UNREGISTERED bounded preflight input planning, not a conditioning runner. + +Only the explicitly called generate_epoch_inputs performs random sampling; it +delegates to conditioning_inputs' single full-shape CUDA draw. The plan and +static rate loader do no sampling, neural execution, plasticity, or I/O writes. +The coordinator owns registration, full seals, graph/overlay authority, device +UUID admission, and every epoch update/checkpoint receipt. +""" +from __future__ import annotations + +import ast +import copy +import hashlib +import io +import json +from pathlib import Path, PurePosixPath +from typing import Any + +import numpy as np + +import conditioning_inputs as inputs +import trial_plan + + +MASTER = 20260930 +BLOCK = 'physiological__PAM15__reinforce_A__rep01' +SCHEMA = 'phase4an-conditioning-preflight-inputs-v1-unregistered' +STATIC = 'experiments/phase4l-olfactory-grounding/' +SOURCE_REL = 'models/male-cns-lif/phase4l.py' +CONFIG_REL = 'models/male-cns-lif/olfactory-grounding-v1/phase4l.yaml' +SOURCE_PINS = { + SOURCE_REL: 'e97d58d8c1f44920d0e953f8f5831ddc86fa0198784e86460e315de4116ac553', + CONFIG_REL: '9b8bcb7e031bf9063033a1854a439a7cd0bf6730301c2d2d02ca99a76f098d6e', + STATIC + 'odor-encoder/frozen-odor-panel.parquet': '907f16a56e024a8e2112b5751e53b0e6037b329fa110098af3d31b119c10a1a0', + STATIC + 'odor-encoder/mapped-glomerular-vectors.npz': 'f057563152e0fa22060080af13019f319e7f36563a775eff829bbc1929c451b0', + STATIC + 'odor-encoder/selection.json': '1b832becf9d08935716fe9ba5e3cd3e619e7d208dfb43c4fa197c969d095f113', + STATIC + 'receptor-mapping/olfactory-mapping.parquet': 'a01d27000533e9844458a4c559cac084f1c34ca20a1104e97e6a0f72b0cc93b1', +} + + +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 build_preflight_plan(): + """Rebuild the whole semantic plan, then select by rule, never outcomes. + + D1/D0/frozen share the two paired input epochs. Every neural epoch starts + fresh; only the externally verified plastic delta can cross its boundary. + This inventory does not enable the complete 68,000-epoch protocol. + """ + full = trial_plan.build_trial_plan(MASTER) + trial_plan.validate_trial_plan(full) + block = full['plan']['blocks'][0] + require(block['block_id'] == BLOCK and block['mode'] == 'physiological' and + block['dan_type'] == 'PAM15' and block['compartment_verbatim'] == "y5B'2a", 'First fixed block changed') + suite = next(s for s in block['suites'] if s['assay'] == 'normal') + paired = next(a for a in suite['arms'] if a['arm'] == 'paired') + first = next(i for i, epoch in enumerate(paired['epochs']) if epoch['us']['requested']) + require(first + 1 < len(paired['epochs']), 'Requested US lacks its following complete epoch') + epochs = copy.deepcopy(paired['epochs'][first:first + 2]) + calls = [] + for arm in ('D1', 'D0', 'frozen'): + for mode in ('reference', 'light'): + previous = None + for position, epoch in enumerate(epochs): + _validate_epoch(epoch) + row_id = f'{arm}--{mode}--epoch-{position:03d}' + calls.append(dict(row_id=row_id, arm=arm, diagnostic_mode=mode, + input_epoch_index=position, epoch_id=epoch['epoch_id'], + full_plan_epoch_index=epoch['epoch_index'], steps=1200, + d_command=0.0 if arm == 'D0' else epoch['D_command'], + frozen=arm == 'frozen', previous_delta_row_id=previous, + neural_state='fresh_rest_zero_ref_zero_spike_queue_zero_APL_queue', + input_sharing='same saved CS and US arrays across arms and diagnostic modes')) + previous = row_id + result = dict(schema=SCHEMA, master=MASTER, block_id=BLOCK, mode='physiological', + dan_type='PAM15', compartment_verbatim="y5B'2a", full_trial_plan=full, + selected_epochs=epochs, calls=calls, total_calls=12, total_ticks=14400, + input_generation_calls=2, input_generation_executed=False, brain_runs_executed=0, + complete_conditioning_authorized=False, training_result_claimed=False, + selection_rule='first normal paired requested US epoch in first fixed block, plus its immediate successor') + result['payload_sha256'] = _sha(result) + return result + + +def validate_preflight_plan(value): + require(type(value) is dict and _canonical(value) == _canonical(build_preflight_plan()), + 'Preflight inventory differs from full deterministic reconstruction') + return True + + +def _validate_epoch(epoch): + require(type(epoch) is dict and epoch.get('steps') == 1200 and epoch.get('duration_ms') == 120 and + epoch.get('initial_neural_state') == 'frozen_reset_including_KC_trace_and_both_delay_queues', + 'Only complete fresh normal 120 ms epochs are in this preflight') + cs, us = epoch.get('cs'), epoch.get('us') + require(type(cs) is dict and cs.get('cue') in ('A', 'B') and cs.get('odor') == trial_plan.ODORS[cs['cue']] and + type(cs.get('seed')) is int and 0 <= cs['seed'] < 2**63 and cs.get('seed_bits') == 63 and + cs.get('start_tick') == 0 and cs.get('stop_tick_exclusive') == 1000, 'CS semantic descriptor changed') + require(type(us) is dict and type(us.get('requested')) is bool and us.get('nominal_slot_tick') == 900 and + us.get('rng_seed') is None and us.get('rng_draws') == 0, 'US must remain a deterministic scheduled pulse') + require((us['requested'] and us.get('tick') == 900 and us.get('request_rule') == 'fixed_single_pulse_all_bound_DAN_targets') or + (not us['requested'] and us.get('tick') is None and us.get('request_rule') == 'none'), 'US timing/request rule changed') + + +def _extract_drive(raw, filename): + """Compile only the SHA-pinned pure pn_drive AST; never import old L.""" + import pandas as pd + require(hashlib.sha256(raw).hexdigest() == SOURCE_PINS[SOURCE_REL], 'Frozen L source SHA mismatch') + nodes = [n for n in ast.parse(raw, filename=filename).body if isinstance(n, ast.FunctionDef) and n.name == 'pn_drive'] + require(len(nodes) == 1 and not nodes[0].decorator_list, 'Unique undecorated pn_drive required') + namespace = {'np': np, 'pd': pd, 'Any': Any} + exec(compile(ast.Module(body=nodes, type_ignores=[]), filename, 'exec'), namespace) + return namespace['pn_drive'], hashlib.sha256(ast.dump(nodes[0], include_attributes=False).encode()).hexdigest() + + +def _rate_tables(selection, panel, vectors, mapping, config, drive): + """Pure static-table derivation, preserving every original PN column.""" + require(selection.get('selected_before_KC_simulation') is True and len(panel) == 12 and + panel.odor_id.is_unique and panel.split.tolist() == ['calibration'] * 6 + ['heldout'] * 6, + 'Frozen static odor panel schema changed') + gloms = selection['glomeruli_order'] + require(type(gloms) is list and len(gloms) == len(set(gloms)) and len(gloms) > 0 and + set(vectors) == set(panel.odor_id), 'Frozen vector/glomerulus inventory mismatch') + encoder = config['odor_encoder'] + require(encoder['max_total_rate_per_glomerulus_hz'] == 400.0 and + encoder['modes'] == ['physiological', 'matched_energy'], 'Frozen rate transform changed') + median = float(np.median(panel.mapped_response_sum)) + require(np.isfinite(median) and median > 0, 'Invalid panel response median') + tables = {} + for cue, name in trial_plan.ODORS.items(): + selected = selection['ODOR_' + cue] + require(selected['odor_name'] == name, 'Selected frozen A/B odor name changed') + odor_id = selected['odor_id'] + positions = np.flatnonzero(panel.odor_id.to_numpy() == odor_id) + require(len(positions) == 1, 'Selected odor ID not unique in frozen panel') + panel_row = panel.iloc[int(positions[0])] + require(panel_row.odor_name == name and panel_row.formal_label == 'ODOR_' + cue, 'Panel/selection label mismatch') + vector = vectors[odor_id] + require(isinstance(vector, np.ndarray) and vector.dtype == np.float32 and vector.shape == (len(gloms),) and + np.isfinite(vector).all() and np.all(vector >= 0), 'Invalid frozen glomerular vector; never clip') + targets, rates, frame = drive(odor_id, vector, mapping, gloms, 'physiological', median, config) + inputs.probabilities_f32(targets, rates) + require(len(targets) > 0 and np.all(rates > 0), 'Empty or nonpositive selected PN drive') + require(frame.neuron_idx.to_numpy(np.int32).tobytes() == targets.tobytes() and + frame.rate_hz.to_numpy(np.float32).tobytes() == rates.tobytes(), 'Original frame/array ordering mismatch') + targets, rates = targets.copy(), rates.copy() + targets.setflags(write=False); rates.setflags(write=False) + tables[cue] = dict(cue=cue, odor_id=str(odor_id), odor_name=name, mode='physiological', + source_panel_row=int(positions[0]), source_panel_split=str(panel_row.split), + targets=targets, rates_hz=rates, identities={ + 'targets': inputs.array_identity(targets), 'rates_hz': inputs.array_identity(rates)}, + glomeruli_order=copy.deepcopy(gloms), mapped_vector_identity=inputs.array_identity(vector)) + return tables, median + + +def load_rate_tables(root): + """Read only the six exact static/source files pinned by frozen AK. + + This does not read any historical neural result or rerun odor selection. + Source files are checked before parsing; the parsed bytes are the same + checked snapshots. Full prior seal/registration authority remains external. + """ + import pandas as pd + import yaml + root = Path(root) + raw = {rel: (root / rel).read_bytes() for rel in SOURCE_PINS} + require(all(hashlib.sha256(value).hexdigest() == SOURCE_PINS[rel] for rel, value in raw.items()), + 'Frozen static encoder/source SHA mismatch') + source, ast_sha = _extract_drive(raw[SOURCE_REL], str(root / SOURCE_REL)) + selection = json.loads(raw[STATIC + 'odor-encoder/selection.json']) + panel = pd.read_parquet(io.BytesIO(raw[STATIC + 'odor-encoder/frozen-odor-panel.parquet'])) + mapping = pd.read_parquet(io.BytesIO(raw[STATIC + 'receptor-mapping/olfactory-mapping.parquet'])) + config = yaml.safe_load(raw[CONFIG_REL]) + with np.load(io.BytesIO(raw[STATIC + 'odor-encoder/mapped-glomerular-vectors.npz']), allow_pickle=False) as values: + require(set(values.files) == set(panel.odor_id) | {'glomeruli'} and + values['glomeruli'].tolist() == selection['glomeruli_order'], 'Frozen NPZ inventory/order mismatch') + vectors = {str(oid): values[oid].copy() for oid in panel.odor_id} + tables, median = _rate_tables(selection, panel, vectors, mapping, config, source) + return dict(tables=tables, audit=dict(schema=SCHEMA, static_sources=copy.deepcopy(SOURCE_PINS), + original_pn_drive_AST_sha256=ast_sha, historical_dynamic_results_read=False, + original_function_called_without_PN_reordering=True, mode='physiological', + median_mapped_response_sum=median, full_seal_verification_owned_by_coordinator=True, + random_draws=0, brain_runs=0, numpy_version=np.__version__, pandas_version=pd.__version__)) + + +def _validate_table(table, cue): + require(type(table) is dict and table.get('cue') == cue and table.get('odor_name') == trial_plan.ODORS[cue] and + table.get('mode') == 'physiological' and type(table.get('odor_id')) is str and table['odor_id'], 'Rate-table cue identity mismatch') + inputs.probabilities_f32(table['targets'], table['rates_hz']) + require(table['identities'] == {key: inputs.array_identity(table[key]) for key in ('targets', 'rates_hz')}, + 'Rate-table arrays changed after authority binding') + + +def generate_epoch_inputs(epoch, tables, dan_targets, *, device, max_uniform_draws, conflict_policy='error'): + """Future explicit sampler, called once per selected input epoch, not arm. + + No brain reset is called. The scheduled US request is identical for D1, + D0, and frozen; D-command and update policy belong to the epoch authority. + The caller must preserve and share these returned bytes across all arms. + """ + _validate_epoch(epoch) + cs = epoch['cs']; table = tables[cs['cue']]; _validate_table(table, cs['cue']) + require(isinstance(dan_targets, np.ndarray) and dan_targets.dtype == np.int32 and dan_targets.ndim == 1 and + len(dan_targets) > 0 and np.all((dan_targets >= 0) & (dan_targets < inputs.N_NEURONS)) and + (len(dan_targets) < 2 or np.all(dan_targets[1:] > dan_targets[:-1])), 'Externally bound DAN globals must be sorted unique int32') + require(conflict_policy in ('error', 'boolean_union'), 'Unknown explicit request composition policy') + cs_result = inputs.generate_cs_requests(table['targets'], table['rates_hz'], seed=cs['seed'], steps=1200, + cs_start_tick=0, cs_stop_tick=1000, device=device, max_uniform_draws=max_uniform_draws) + nodes = dan_targets.copy() if epoch['us']['requested'] else np.empty(0, np.int32) + us = {'tick': np.full(len(nodes), 900, np.int32), 'neuron_idx': nodes} + for value in us.values(): value.setflags(write=False) + composed = inputs.merge_requests(cs_result['requests'], us, steps=1200, conflict_policy=conflict_policy) + return dict(requests=composed['requests'], cs_requests=cs_result['requests'], us_requests=us, + source_mask=composed['source_mask'], audit=dict(schema=SCHEMA, epoch_id=epoch['epoch_id'], + CS_sampler=cs_result['audit'], composition=composed['audit'], + DAN_targets_identity=inputs.array_identity(dan_targets), requested_US=epoch['us']['requested'], + scheduled_US_tick=epoch['us']['tick'], CS_shared_across_arms=True, US_shared_across_arms=True, + US_rng_draws=0, semantic_seed_never_passed_to_brain_reset=True, brain_reset_called=False, + source_authority_and_saved_array_rehash_owned_by_coordinator=True)) + + +def derive_case_spec(original, row, epoch, rate_table, *, brain_reset_seed, budget_root, output_budget_bytes): + """Copy an externally authenticated wholegraph spec with explicit changes. + + ``seed`` is an independent conservative 31-bit reset-control value, not a + conversion of the semantic input seed. The inherited deterministic AJ + runtime consumes saved requests, not this input RNG. No checkpoint carry + is admitted: the next epoch carries only its separately verified delta. + """ + _validate_epoch(epoch); _validate_table(rate_table, epoch['cs']['cue']) + require(type(original) is dict and original.get('kind') == 'fullgraph' and original.get('dt_ms') == .1 and + original.get('apl_delay_steps') == 18 and original.get('precision') == 'float64' and + original.get('apl_precision') == 'float32' and original.get('record_history') is False and + original.get('within_trial_weight_updates', False) is False, 'Anchored fresh D18 fullgraph spec required') + require(not any('checkpoint' in str(k).lower() or k in ('initial_state', 'resume', 'continuation', 'global_tick_offset') for k in original), + 'No neural checkpoint continuation in this two-epoch preflight') + require(type(brain_reset_seed) is int and 0 <= brain_reset_seed < 2**31, 'Independent conservative 31-bit brain reset seed required') + require(type(row) is dict and row.get('arm') in ('D1', 'D0', 'frozen') and row.get('diagnostic_mode') in ('reference', 'light') and + row.get('epoch_id') == epoch['epoch_id'] and row.get('steps') == 1200 and + row.get('frozen') is (row['arm'] == 'frozen') and + row.get('d_command') == (0.0 if row['arm'] == 'D0' else epoch['D_command']), 'Row/epoch command mismatch') + require(type(row.get('input_epoch_index')) is int and row['input_epoch_index'] in (0, 1) and + row.get('row_id') == f"{row['arm']}--{row['diagnostic_mode']}--epoch-{row['input_epoch_index']:03d}", 'Unsafe row identity') + require(type(budget_root) is str and budget_root.startswith('/') and not budget_root.startswith('//') and + '..' not in PurePosixPath(budget_root).parts and + type(output_budget_bytes) is int and output_budget_bytes > 0, 'Explicit absolute output budget required') + old = PurePosixPath(original['budget_root']); new = PurePosixPath(budget_root) + require(old.is_absolute() and not new.is_relative_to(old) and not old.is_relative_to(new), 'Budget cannot overlap old experiment') + result = copy.deepcopy(original) + removed = {key: result.pop(key) for key in ('validation_cohort', 'odor_index', 'expected_external_events', 'training') if key in result} + cs = epoch['cs']; rates = rate_table['rates_hz'] + result.update(name=row['row_id'], steps=1200, duration_ms=120.0, max_steps=1200, + budget_root=budget_root, output_budget_bytes=output_budget_bytes, seed=brain_reset_seed, + cs_input_seed=cs['seed'], cs_input_seed_bits=63, + seed_scope='seed is separate reset control; full cs_input_seed used only by saved-input sampler', + mode='physiological', odor_id=rate_table['odor_id'], odor_name=rate_table['odor_name'], + source_panel_row=rate_table['source_panel_row'], source_panel_split=rate_table['source_panel_split'], + rate_sum_hz=float(rates.sum()), target_count=len(rate_table['targets']), + expected_CS_requests=float(np.sum(inputs.probabilities_f32(rate_table['targets'], rates), dtype=np.float64) * 1000), + CS_window=[0, 1000], within_trial_weight_updates=False, + outer_protocol='conditioning_interface_preflight_not_full_conditioning', + outer_epoch_id=epoch['epoch_id'], outer_arm=row['arm']) + changes = {k: dict(present_before=k in original, before=copy.deepcopy(original.get(k)), after=copy.deepcopy(v)) + for k, v in result.items() if k not in original or _canonical(v) != _canonical(original[k])} + allowed = {'name', 'steps', 'duration_ms', 'max_steps', 'budget_root', 'output_budget_bytes', 'seed', + 'cs_input_seed', 'cs_input_seed_bits', 'seed_scope', 'mode', 'odor_id', 'odor_name', 'source_panel_row', + 'source_panel_split', 'rate_sum_hz', 'target_count', 'expected_CS_requests', 'CS_window', + 'within_trial_weight_updates', 'outer_protocol', 'outer_epoch_id', 'outer_arm'} + require(set(changes) <= allowed and all(_canonical(result[k]) == _canonical(v) for k, v in original.items() + if k not in allowed and k not in removed), 'Unexpected frozen spec change') + return result, dict(schema=SCHEMA, explicit_field_changes=changes, removed_stale_input_metadata=removed, + original_spec_sha256=_sha(original), derived_spec_sha256=_sha(result), + graph_APL_model_coefficients_and_other_fields_unchanged=True, + independent_brain_reset_seed=brain_reset_seed, semantic_input_seed=cs['seed'], + neural_state_between_epochs='fresh reset; only plastic delta may carry via separate verified authority', + D_and_update_commands_owned_by_epoch_authority=True, random_draws=0, brain_runs=0, + outer_learning_pass_claimed=False, full_conditioning_authorized=False) diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_prepare.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_prepare.py new file mode 100644 index 0000000..1172dad --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_prepare.py @@ -0,0 +1,197 @@ +"""Exclusive CPU preparation for the finite AO conditioning-interface gate. + +This is a two-epoch installation test, not acquisition, retention or a learning +candidate. Source authority and the complete 12-call plan precede CUDA sampling. +Generated requests and learned snapshots receive separate execution receipts; +their not-yet-existing hashes are never invented in the registration. +""" +from __future__ import annotations + +import json +from pathlib import Path +import shutil +import subprocess +import sys + +import numpy as np + +import engineering_prepare as P +import runtime_light as R +import biological_contract as B +import conditioning_preflight_inputs as I + +ROOT = Path('/data/digital-drosophila') +MODEL = Path(__file__).resolve().parent +EXP = ROOT / 'experiments/phase4ao-conditioning-interface' +AN = ROOT / 'experiments/phase4an-engineering-validation' +AN_REG = '7317441126f191ccc6e6d28438870b3921537ebe282a6acb163716a690057945' +AN_MANIFEST = 'e8952a48cd3ea6a93c811a1183a1f7b79e19646299924e7688e5f393737a9aa2' +AN_FINAL = '16630d0d666a2467c56b647d021cbdb627a553795f6afcd7f7eb496a2737d815' +PREP_COMMIT = 'c187d7d5a6047f183467e3d860e2529575dcda97' +DRAFT_INVENTORY_SHA = '2c72d47d09b4ad1949dc6cfdc170f64847ff75f95ded5d20b421fb5a75e0eb84' +BUDGET = dict(output_budget_bytes=4 * 1024**3, minimum_free_reserve_bytes=10 * 1024**3, + max_uniform_draws=2_000_000, child_timeout_seconds=1800) +require, read, save, digest = P.require, P.read, P.save, P.digest + + +def same(a, b): + return json.dumps(a, sort_keys=True, allow_nan=False) == json.dumps(b, sort_keys=True, allow_nan=False) + + +def verify_pins(pins): + for name, sha in pins.items(): + path = Path(name) + require(path.is_absolute() and path.is_file() and not path.is_symlink() and + digest(path) == sha, 'Pinned file changed: ' + name) + + +def save_npz(path, **arrays): + with Path(path).open('xb') as stream: + np.savez_compressed(stream, **arrays) + + +def source_inventory(): + # Explicitly snapshot the current lab's text sources, including this new + # coordinator and its tests. Existing 21/20 files are never rewritten. + paths = sorted([*MODEL.glob('*.py'), *MODEL.glob('*.md')]) + required = ('conditioning_epoch.py', 'conditioning_preflight_inputs.py', + 'conditioning_preflight_compare.py', 'conditioning_preflight_prepare.py', + 'conditioning_preflight_workflow.py', 'test_conditioning_preflight_prepare.py', + 'CONDITIONING_INTERFACE_PROTOCOL.md') + require(all(MODEL / n in paths for n in required), 'Missing complete coordinator/test/protocol source') + require(not any(p.is_symlink() for p in paths), 'Linked source is forbidden') + return {str(p): digest(p) for p in paths} + + +def plan_descriptor(full): + return {k: v for k, v in full.items() if k not in ('full_trial_plan', 'payload_sha256')} + + +def prepare(): + require(sys.dont_write_bytecode, 'Use scientific Python -B') + require(EXP.is_dir() and not any(EXP.iterdir()) and not EXP.is_symlink(), + 'Preparation requires the explicitly created empty AO directory; never overwrite') + require(digest(AN / 'manifest.json') == AN_MANIFEST and digest(AN / 'audit/final-status.json') == AN_FINAL, + 'Externally anchored AN controls changed') + an = P.read_registration(AN_REG) + helper = P.load_module('phase4ao_original_seal_helper', ROOT / P.AI_HELPER_REL, P.AI_HELPER_SHA) + an_seal = helper.verify_seal(AN) + am_seal = P.verify_authority(an['anchors']) + old = P.verify_old_integrity(an) + full = I.build_preflight_plan() + rates = I.load_rate_tables(ROOT) + populations, biology = B.load_bound_tables(ROOT, an['input_sha256']) + pins = dict(an['input_sha256']) + pins.update({str(ROOT / k): v for k, v in I.SOURCE_PINS.items()}) + pins.update(biology['input_sha256']) + pins.update({str(AN / 'manifest.json'): AN_MANIFEST, str(AN / 'audit/final-status.json'): AN_FINAL, + str(AN / 'protocol/preregistered.json'): AN_REG}) + bridge = ROOT / 'models/male-cns-lif/wholebrain-plastic-interface-v1/eligibility_bridge.py' + phasej = ROOT / 'models/male-cns-lif/mb-physiology-v1/phase4j.py' + pins[str(bridge)] = '174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61' + pins[str(phasej)] = 'abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e' + original_path = Path(an['selected_case']['original_dir']) / 'spec.json' + require(str(original_path) in pins, 'Original graph spec not anchored by AN') + original = read(original_path) + previous = MODEL / 'draft-source-inventory.json' + require(digest(previous) == DRAFT_INVENTORY_SHA, 'Previously delivered 20-file source inventory changed') + inventory = read(previous) + require(len(inventory['files']) == 20, 'Previous scientific snapshot must contain 20 files') + pins[str(previous)] = DRAFT_INVENTORY_SHA + for entry in inventory['files']: + require(Path(entry['path']).name == entry['path'], 'Nonlocal scientific source inventory entry') + pins[str(MODEL / entry['path'])] = entry['sha256'] + zero = an['overlays']['zero'] + _, overlay, metadata, _ = R.load_overlay(zero['overlay_file'], zero['overlay_manifest']) + require(not np.any(overlay['plastic_delta_mv'].view(np.uint32)), 'Initial delta must be exact positive zero') + kc = np.load(Path(original['apl_bundle_dir']) / 'kc_globals.npy', allow_pickle=False) + require(np.array_equal(kc, populations['Kenyon_Cell']), 'Canonical metadata and APL KC memberships disagree') + import pandas as pd + candidate = pd.read_parquet(ROOT / 'experiments/phase4h-dopamine-plasticity/plastic-mask/plastic-edge-candidates.parquet') + candidate = candidate.set_index('edge_id').loc[overlay['edge_id']].reset_index() + require(candidate.edge_id.to_numpy().tobytes() == overlay['edge_id'].tobytes() and + candidate.target_compartment_verbatim.notna().all(), 'Complete mask compartment mapping is not exact') + verify_pins(pins) + sources = source_inventory() + code_commit = subprocess.run(['git', '-C', str(ROOT), 'rev-parse', 'HEAD'], check=True, + capture_output=True, text=True).stdout.strip() + require(P.sha(code_commit + '0' * 24), 'Invalid Git code version') + for path, sha in sources.items(): + relative = Path(path).relative_to(ROOT).as_posix() + blob = subprocess.run(['git', '-C', str(ROOT), 'show', code_commit + ':' + relative], + check=True, capture_output=True).stdout + require(__import__('hashlib').sha256(blob).hexdigest() == sha, 'Registered source is not the committed Git blob: ' + relative) + for name in ('protocol/source-snapshot', 'inputs/static', 'cases', 'runs', 'updates', 'receipts', 'logs', 'audit', 'regression'): + (EXP / name).mkdir(parents=True) + for path in sources: + shutil.copyfile(path, EXP / 'protocol/source-snapshot' / Path(path).name) + save(EXP / 'regression/AN-seal-before.json', an_seal) + save(EXP / 'regression/AM-seals-before.json', am_seal) + save(EXP / 'regression/old-integrity-before.json', old) + save(EXP / 'inputs/static/biology-audit.json', biology) + save(EXP / 'inputs/static/rates-audit.json', rates['audit']) + save_npz(EXP / 'inputs/static/populations.npz', **populations) + save_npz(EXP / 'inputs/static/candidate-map.npz', edge_id=candidate.edge_id.to_numpy(np.int64), + source_neuron_idx=candidate.source_neuron_idx.to_numpy(np.int32), + target_neuron_idx=candidate.target_neuron_idx.to_numpy(np.int32), + target_compartment_verbatim=candidate.target_compartment_verbatim.to_numpy(dtype=str)) + table_metadata = {} + for cue, table in rates['tables'].items(): + save_npz(EXP / f'inputs/static/{cue}-rates.npz', targets=table['targets'], rates_hz=table['rates_hz']) + table_metadata[cue] = {k: v for k, v in table.items() if k not in ('targets', 'rates_hz')} + save(EXP / 'inputs/static/rate-tables.json', table_metadata) + save(EXP / 'protocol/full-trial-plan.json', full['full_trial_plan']) + descriptor = plan_descriptor(full) + cases = {} + reset_seed = original['seed'] + for row in descriptor['calls']: + epoch = descriptor['selected_epochs'][row['input_epoch_index']] + case = EXP / 'cases' / row['row_id'] + case.mkdir() + spec, derivation = I.derive_case_spec(original, row, epoch, rates['tables'][epoch['cs']['cue']], + brain_reset_seed=reset_seed, budget_root=str(EXP), output_budget_bytes=BUDGET['output_budget_bytes']) + R.validate_spec(spec, case) + save(case / 'spec.json', spec) + save(case / 'derivation.json', derivation) + cases[row['row_id']] = str(case) + derived = {str(p): digest(p) for parent in ('inputs/static', 'cases', 'protocol/source-snapshot', 'regression') + for p in (EXP / parent).rglob('*') if p.is_file()} + derived[str(EXP / 'protocol/full-trial-plan.json')] = digest(EXP / 'protocol/full-trial-plan.json') + reg = dict(schema='phase4ao-conditioning-interface-registration-v1', plan=descriptor, + full_plan_payload_sha256=full['payload_sha256'], full_trial_plan_file_sha256=derived[str(EXP / 'protocol/full-trial-plan.json')], + source_sha256=sources, input_sha256=pins, prepared_file_sha256=derived, cases=cases, + initial_overlay=zero, original_spec=str(original_path), brain_reset_seed=reset_seed, + bridge_source=str(bridge), phase4j_source=str(phasej), budget=BUDGET, + git_parent_commit=PREP_COMMIT, git_code_commit=code_commit, committed_source_blobs_verified=len(sources), AN_registration_sha256=AN_REG, + generation=dict(device='cuda:0', GPU_UUID='GPU-79f3a85d-adce-7833-7003-6488140ca596', + conflict_policy='error', highbit_probe_seed=descriptor['selected_epochs'][0]['cs']['seed'], + highbit_comparison='full63bit versus low32bit, one fixed 1200xPN matrix each; never a seed search', + original_L_sampling_AST='only rate tensor, private generator and full-shape forced-spike expression', + trace_backend='torch', actual_input_files='created only after external registration SHA recorded'), + scope=dict(epoch_boundary_updates=True, within_trial_updates=False, full_conditioning=False, + candidate_generated=False, learning_success_claimed=False, neural_state_carried=False, + delta_carried=True, independent_model_branch=True), + success_gate='12 complete calls; exact reference/light events and final state/queues; exact D0/frozen identity; real D1 delta installed in epoch 2; nonzero D1 update coverage required', + missing_nonzero_coverage='INCOMPLETE, do not choose another input', + failure_policy='preserve all partials; no automatic run retry or result-selected seed', + retention_supported=False, future_hashes_fabricated=False, new_brain_runs_during_preparation=0) + verify_pins({**pins, **sources, **derived}) + save(EXP / 'protocol/preregistered.json', reg) + return digest(EXP / 'protocol/preregistered.json') + + +def read_registration(expected): + path = EXP / 'protocol/preregistered.json' + require(P.sha(expected) and digest(path) == expected, 'Registration differs from external anchor') + reg = read(path) + require(reg['schema'] == 'phase4ao-conditioning-interface-registration-v1' and + reg['AN_registration_sha256'] == AN_REG and reg['budget'] == BUDGET and + reg['plan']['total_calls'] == 12 and reg['plan']['total_ticks'] == 14400, + 'Unexpected finite registration') + verify_pins({**reg['source_sha256'], **reg['input_sha256'], **reg['prepared_file_sha256']}) + return reg + + +if __name__ == '__main__': + require(sys.argv[1:] == ['prepare'], 'Only exclusive prepare is supported') + print('REGISTRATION_SHA256', prepare(), flush=True) diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_workflow.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_workflow.py new file mode 100644 index 0000000..717e232 --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/conditioning_preflight_workflow.py @@ -0,0 +1,474 @@ +"""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) diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_epoch.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_epoch.py new file mode 100644 index 0000000..516f45f --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_epoch.py @@ -0,0 +1,298 @@ +"""Tiny temporary CPU fixtures, never scientific events or a real brain run.""" +import copy +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +import numpy as np + +import conditioning_epoch as m +import conditioning_update as update +from test_conditioning_update import BRIDGE, SOURCE + + +def save(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, sort_keys=True, allow_nan=False), encoding='utf-8') + + +class TinyFixture: + def __init__(self, root): + self.root = root + self.inputs = root / 'inputs'; self.inputs.mkdir() + self.case = self.inputs / 'case'; self.case.mkdir() + self.graph = self.inputs / 'graph'; self.graph.mkdir() + self.apl = self.inputs / 'apl'; self.apl.mkdir() + self.checkpoint = root / 'runtime'; self.checkpoint.mkdir() + self.authority_dir = root / 'authority'; self.authority_dir.mkdir() + self.output_parent = root / 'outputs'; self.output_parent.mkdir() + self.authority_path = self.authority_dir / 'epoch.json' + self.mask = np.array([1, 3], np.int64) + self.kc = np.array([1, 3], np.int32) + self.base = np.array([.275, .55], np.float32) + self.before = np.zeros(2, np.float32) + graph = dict(nodes=np.arange(5, dtype=np.int32), edge_id=np.arange(6, dtype=np.int64), + src=np.array([0, 1, 1, 3, 3, 4], np.int32), dst=np.array([1, 2, 4, 2, 4, 0], np.int32), + signed_raw=np.ones(6, np.int64), ptr=np.array([0, 1, 3, 3, 5, 6], np.int64), weights=np.full(6, .000275)) + for name, array in graph.items(): + np.save(self.graph / (name + '.npy'), array) + for name, array in dict(input_matrix=np.zeros((2, 5), np.float32), + output_matrix=np.zeros((2, 2), np.float32), kc_globals=self.kc, apl_globals=np.array([0, 4], np.int32)).items(): + np.save(self.apl / (name + '.npy'), array) + np.save(self.inputs / 'kc.npy', self.kc) + np.save(self.inputs / 'mask.npy', self.mask) + np.save(self.inputs / 'base.npy', self.base) + np.savez_compressed(self.inputs / 'mapping.npz', edge_id=self.mask, + source_neuron_idx=graph['src'][self.mask], target_neuron_idx=graph['dst'][self.mask], + target_compartment_verbatim=np.array(["y5B'2a", 'a1'])) + np.savez_compressed(self.inputs / 'before.npz', edge_id=self.mask, plastic_delta_mv=self.before) + save(self.inputs / 'coeff.json', {'synthetic': True}) + source_root = Path(m.__file__).resolve().parent + source_files = dict(runtime_light_source=source_root / 'runtime_light.py', + runtime_continuation_source=source_root / 'runtime_continuation.py', + conditioning_update_source=source_root / 'conditioning_update.py', + eligibility_bridge_source=BRIDGE.resolve(), phase4j_source=SOURCE.resolve(), + am_overlay_source=source_root.parent / 'wholebrain-plastic-interface-v1/overlay.py') + metadata = dict(schema='phase4an-overlay-v1', overlay_name='zero', + overlay_sha256=m._digest(self.inputs / 'before.npz'), mask_path=str(self.inputs / 'mask.npy'), + mask_sha256=m._digest(self.inputs / 'mask.npy'), base_candidate_path=str(self.inputs / 'base.npy'), + base_candidate_sha256=m._digest(self.inputs / 'base.npy'), aj_model_dir=str(self.graph), + am_overlay_path=str(source_files['am_overlay_source']), am_overlay_sha256=m._digest(source_files['am_overlay_source'])) + save(self.inputs / 'before.json', metadata) + spec = dict(name='synthetic', steps=4, max_steps=4, dt_ms=.1, duration_ms=.4, + apl_delay_steps=18, record_history=False, within_trial_weight_updates=False, + graph_dir=str(self.graph), apl_bundle_dir=str(self.apl), coefficients_path=str(self.inputs / 'coeff.json')) + save(self.case / 'spec.json', spec) + np.savez_compressed(self.case / 'requests.npz', tick=np.array([0], np.int32), neuron_idx=np.array([1], np.int32)) + state = dict(step=np.array([0], np.int32), neuron_idx=np.array([1], np.int32), + counts=np.array([0, 1, 0, 0, 0], np.int32), v=np.full(5, -.052), g=np.zeros(5), ref=np.zeros(5, np.int32)) + np.savez_compressed(self.checkpoint / 'state.npz', **state) + (self.checkpoint / 'overlay-input.npz').write_bytes((self.inputs / 'before.npz').read_bytes()) + (self.checkpoint / 'overlay-manifest.json').write_bytes((self.inputs / 'before.json').read_bytes()) + summary = dict(nodes=5, edges=6, diagnostic_mode='light', effective_weights_raw_sha256='e' * 64) + save(self.checkpoint / 'summary.json', summary) + provenance = dict(input_sha256={str(path): m._digest(path) for path in + (self.case / 'spec.json', self.case / 'requests.npz', self.inputs / 'before.npz', self.inputs / 'before.json')}) + save(self.checkpoint / 'runtime-provenance.json', provenance) + save(self.checkpoint / 'overlay-audit.json', dict(delta_raw_sha256=self.raw(self.before), effective_weights_raw_sha256='e' * 64)) + paths = dict(spec=self.case / 'spec.json', requests=self.case / 'requests.npz', + overlay_before=self.inputs / 'before.npz', overlay_manifest_before=self.inputs / 'before.json', + state=self.checkpoint / 'state.npz', kc=self.inputs / 'kc.npy', canonical_kc=self.apl / 'kc_globals.npy', + candidate_map=self.inputs / 'mapping.npz', mask=self.inputs / 'mask.npy', base_candidate=self.inputs / 'base.npy', **source_files) + self.value = dict(schema=m.SCHEMA, epoch_id='synthetic/epoch-0', diagnostic_mode='light', + command=dict(steps=4, compartment="y5B'2a", us_tick=1, d_command=1., frozen=False, trace_backend='numpy'), + checkpoint_dir=str(self.checkpoint), checkpoint_manifest_sha256=None, + files={role: dict(path=str(path), sha256=m._digest(path)) for role, path in paths.items()}, + frozen_inputs={str(path): m._digest(path) for path in [ + *(self.graph / name for name in m.GRAPH_FILES), *(self.apl / name for name in m.APL_FILES), self.inputs / 'coeff.json']}) + self.reseal() + + @staticmethod + def raw(array): + return __import__('hashlib').sha256(array.tobytes()).hexdigest() + + def reseal(self): + files = {path.name: m._digest(path) for path in self.checkpoint.iterdir() if path.name != 'checkpoint-manifest.json'} + save(self.checkpoint / 'checkpoint-manifest.json', dict(files_sha256=files)) + self.value['checkpoint_manifest_sha256'] = m._digest(self.checkpoint / 'checkpoint-manifest.json') + for entry in self.value['files'].values(): + entry['sha256'] = m._digest(entry['path']) + for path in self.value['frozen_inputs']: + self.value['frozen_inputs'][path] = m._digest(path) + save(self.authority_path, self.value) + + def load_checkpoint(self, directory, pin): + self.asserted_loads += 1 + directory = Path(directory) + m._require(m._digest(directory / 'checkpoint-manifest.json') == pin, 'synthetic seal pin mismatch') + manifest = m._read(directory / 'checkpoint-manifest.json') + actual = {path.name: m._digest(path) for path in directory.iterdir() if path.name != 'checkpoint-manifest.json'} + m._require(manifest['files_sha256'] == actual, 'synthetic sealed inventory mismatch') + return dict(fresh=True, global_tick_offset=0, steps=4, summary=m._read(directory / 'summary.json'), + provenance=m._read(directory / 'runtime-provenance.json'), manifest=manifest, + state=m._npz(directory / 'state.npz', ('step', 'neuron_idx', 'counts', 'v', 'g', 'ref')), + overlay=m._read(directory / 'overlay-audit.json')) + + def load_overlay(self, file, manifest): + meta = m._read(manifest) + m._require(meta['overlay_sha256'] == m._digest(file), 'synthetic overlay SHA mismatch') + value = m._npz(file, ('edge_id', 'plastic_delta_mv')) + value['base_candidate_mv'] = np.load(meta['base_candidate_path'], allow_pickle=False) + if meta['overlay_name'] == 'learned': + expected = {'schema', 'overlay_name', 'overlay_sha256', 'mask_path', 'mask_sha256', + 'base_candidate_path', 'base_candidate_sha256', 'aj_model_dir', 'am_overlay_path', 'am_overlay_sha256', + 'parent_snapshot_sha256', 'eligibility_v2_source_sha256'} + m._require(set(meta) == expected, 'learned manifest fields differ from real runtime contract') + return None, value, meta, {} + + def run(self, out=None): + self.asserted_loads = 0 + parsed = m._authority(self.authority_path, m._digest(self.authority_path)) + c = SimpleNamespace(load_checkpoint=self.load_checkpoint) + r = SimpleNamespace(PINS={}, validate_spec=lambda value, _: value, + load_overlay=self.load_overlay, _raw=self.raw) + return m._coordinate(*parsed, self.output_parent / 'result' if out is None else out, + (c, r, update), (5, 6, 2, 2)) + + +class EpochPostprocessing(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.f = TinyFixture(Path(self.tmp.name).resolve()) + + def tearDown(self): + self.tmp.cleanup() + + def test_exclusive_complete_output_and_distinct_parent_hashes(self): + chain = self.f.run() + output = self.f.output_parent / 'result' + self.assertEqual({p.name for p in output.iterdir()}, {'delta-after.npz', 'overlay-manifest.json', 'update-receipt.json', 'chain-receipt.json'}) + meta = m._read(output / 'overlay-manifest.json') + self.assertEqual(meta['parent_snapshot_sha256'], self.f.value['files']['overlay_before']['sha256']) + self.assertNotEqual(meta['parent_snapshot_sha256'], self.f.value['checkpoint_manifest_sha256']) + self.assertEqual(chain['before_overlay_manifest_sha256'], self.f.value['files']['overlay_manifest_before']['sha256']) + self.assertEqual(chain['source_neural_checkpoint_manifest_sha256'], self.f.value['checkpoint_manifest_sha256']) + self.assertEqual(self.f.asserted_loads, 2) + self.assertFalse(chain['weights_installed']) + self.assertFalse(chain['registration_authority_independently_proven']) + after = m._npz(output / 'delta-after.npz', ('edge_id', 'plastic_delta_mv')) + self.assertLess(after['plastic_delta_mv'][0], 0) + self.assertEqual(after['plastic_delta_mv'][1].view(np.uint32), 0) + with self.assertRaisesRegex(ValueError, 'Output must be new'): + self.f.run() + + def test_every_external_sha_before_processing(self): + with self.assertRaisesRegex(ValueError, 'Authority differs'): + m._authority(self.f.authority_path, '0' * 64) + with (self.f.inputs / 'mapping.npz').open('ab') as stream: + stream.write(b'changed') + with self.assertRaisesRegex(ValueError, 'SHA mismatch'): + self.f.run() + self.assertFalse((self.f.output_parent / 'result').exists()) + + def test_sealed_event_substitution_and_unbound_case_are_rejected(self): + path = self.f.checkpoint / 'runtime-provenance.json' + provenance = m._read(path) + provenance['input_sha256'][str(self.f.case / 'requests.npz')] = 'f' * 64 + save(path, provenance) + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'did not bind authorized requests'): + self.f.run() + + def test_changed_runtime_overlay_cannot_match_only_its_own_seal(self): + (self.f.checkpoint / 'overlay-input.npz').write_bytes(b'different') + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'used a different overlay'): + self.f.run() + + def test_candidate_endpoint_and_canonical_kc_must_really_match(self): + path = self.f.inputs / 'mapping.npz' + mapping = m._npz(path, ('edge_id', 'source_neuron_idx', 'target_neuron_idx', 'target_compartment_verbatim')) + mapping['target_neuron_idx'][0] = 4 + np.savez_compressed(path, **mapping) + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'endpoints differ'): + self.f.run() + mapping['target_neuron_idx'][0] = 2 + np.savez_compressed(path, **mapping) + np.save(self.f.inputs / 'kc.npy', np.array([1, 4], np.int32)) + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'KC differs'): + self.f.run() + + def test_command_steps_and_full_frozen_inventory_required(self): + self.f.value['command']['steps'] = 5 + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'Command steps differ'): + self.f.run() + self.f.value['command']['steps'] = 4 + del self.f.value['frozen_inputs'][str(self.f.graph / 'signed_raw.npy')] + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'authority is incomplete'): + self.f.run() + + def test_protected_directories_and_existing_output_fail_before_write(self): + for path in (self.f.checkpoint / 'new', self.f.inputs / 'new', self.f.graph / 'new', + self.f.authority_dir / 'new', self.f.checkpoint): + with self.subTest(path=path), self.assertRaisesRegex(ValueError, 'Output must be new'): + self.f.run(path) + + def test_partial_is_preserved_and_not_retryable(self): + with patch.object(self.f, 'load_overlay', side_effect=[ + self.f.load_overlay(self.f.inputs / 'before.npz', self.f.inputs / 'before.json'), + ValueError('synthetic output validation failed')]): + with self.assertRaisesRegex(ValueError, 'output validation'): + self.f.run() + output = self.f.output_parent / 'result' + self.assertTrue((output / 'postprocess-error.json').is_file()) + self.assertTrue((output / 'delta-after.npz').is_file()) + self.assertFalse((output / 'chain-receipt.json').exists()) + with self.assertRaisesRegex(ValueError, 'Output must be new'): + self.f.run() + + def test_actual_pinned_dependencies_load_without_gpu(self): + _, paths, pins, _, _ = m._authority(self.f.authority_path, m._digest(self.f.authority_path)) + c, r, u = m._production_dependencies(paths, pins) + self.assertEqual(Path(r.__file__).resolve(), paths['runtime_light_source']) + self.assertEqual(c.N, 166700) + self.assertEqual(u.BRIDGE_SHA256, update.BRIDGE_SHA256) + for role in m.SOURCE_PINS: + self.assertEqual(m._digest(paths[role]), m.SOURCE_PINS[role]) + + def test_production_dimensions_cannot_be_set_by_authority(self): + self.assertEqual(m.DIMENSIONS, (166700, 25582938, 4064, 14232)) + self.f.value['small_test_mode'] = True + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'fields/schema'): + m.postprocess_epoch(self.f.authority_path, m._digest(self.f.authority_path), self.f.output_parent / 'result') + + def test_public_entry_rejects_tiny_population_even_with_valid_fixture_seal(self): + self.f.asserted_loads = 0 + c = SimpleNamespace(load_checkpoint=self.f.load_checkpoint) + r = SimpleNamespace(PINS={}, validate_spec=lambda value, _: value, + load_overlay=self.f.load_overlay, _raw=self.f.raw) + with patch.object(m, '_production_dependencies', return_value=(c, r, update)): + with self.assertRaisesRegex(ValueError, 'Runtime population/mode'): + m.postprocess_epoch(self.f.authority_path, m._digest(self.f.authority_path), self.f.output_parent / 'result') + self.assertFalse((self.f.output_parent / 'result').exists()) + + def test_authorized_event_file_cannot_be_an_unsealed_copy(self): + other = self.f.inputs / 'state-copy.npz' + other.write_bytes((self.f.checkpoint / 'state.npz').read_bytes()) + self.f.value['files']['state'] = dict(path=str(other), sha256=m._digest(other)) + self.f.reseal() + with self.assertRaisesRegex(ValueError, 'Events must come from this checkpoint'): + self.f.run() + + def test_late_input_change_preserves_partial_but_never_complete_chain(self): + original = self.f.load_checkpoint + calls = 0 + def late_change(directory, pin): + nonlocal calls + calls += 1 + result = original(directory, pin) + if calls == 2: + with (self.f.graph / 'weights.npy').open('ab') as stream: + stream.write(b'changed after outputs were written') + return result + with patch.object(self.f, 'load_checkpoint', side_effect=late_change): + with self.assertRaisesRegex(ValueError, 'SHA mismatch'): + self.f.run() + output = self.f.output_parent / 'result' + self.assertTrue((output / 'postprocess-error.json').is_file()) + self.assertFalse((output / 'chain-receipt.json').exists()) + + def test_zero_d_and_frozen_return_identity_without_installation(self): + for index, (d, frozen) in enumerate(((0., False), (1., True))): + self.f.value['command'].update(d_command=d, frozen=frozen) + self.f.reseal() + self.f.run(self.f.output_parent / str(index)) + value = m._npz(self.f.output_parent / str(index) / 'delta-after.npz', ('edge_id', 'plastic_delta_mv')) + self.assertEqual(value['plastic_delta_mv'].tobytes(), self.f.before.tobytes()) + + +if __name__ == '__main__': + unittest.main() diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_compare.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_compare.py new file mode 100644 index 0000000..6d60ee8 --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_compare.py @@ -0,0 +1,308 @@ +"""Synthetic CPU array/receipt mutation tests; no files or neural execution.""" +import copy +import hashlib +import json +import unittest +from unittest import mock + +import numpy as np + +import conditioning_preflight_compare as P +from test_engineering_compare import synthetic_parts + + +def rehash(receipt): + receipt["payload_sha256"] = P._sha({k: v for k, v in receipt.items() if k != "payload_sha256"}) + + +def pair(*, active=True): + reference = synthetic_parts([64], active=active)[0] + light = synthetic_parts([64], active=active, traces=False, queues=False)[0] + for record, mode in ((reference, "reference"), (light, "light")): + summary = record["summary"] + summary.update(edges=6, explicit_steps_executed=64, within_trial_weight_updates=False, + hidden_early_stop=False, entire_neural_trial_executed=True, record_history=False, + state_units="SI volts", first_failure_tick=-1, counts_match_sparse_events=True, + total_spikes=len(record["state"]["step"]), requested=len(record["requests"]), + accepted_requests=int(record["requests"][:, 3].sum()), rejected_requests=int(record["requests"][:, 4].sum()), + full_APL_history_recorded=mode == "reference", full_tick_state_hashes_recorded=mode == "reference", + full_queue_hashes_recorded=mode == "reference", full_sequence_bitwise_audit_performed=mode == "reference", + arriving_edges=sum(row["count"] for row in reference["queues"]), + accepted_arriving_edges=sum(row["accepted_count"] for row in reference["queues"]), + input_sha256={"base_weights.npy": "a" * 64, "coefficients.json": "b" * 64, + "requests.npz": "c" * 64, "spec.json": "d" * 64}) + record["terminal_origin"] = "recorded_terminal_queues" + return reference, light, reference["requests"][:, :2].astype(np.int32) + + +def delta_fixture(): + before = np.zeros(P.CANDIDATE_COUNT, np.float32) + after = before.copy() + after[1] = np.float32(-.001) + receipt = dict(schema="phase4an-conditioning-update-draft-v1", + status="DRAFT-COMPUTED-EPOCH-UPDATE-NOT-INSTALLED", + source_pins={"eligibility_bridge_sha256": P.ELIGIBILITY_BRIDGE_SHA256, "phase4j_sha256": P.PHASE4J_SHA256}, + input_arrays={"delta_before": P._evidence(before)}, outputs={"delta_after": P._evidence(after)}, + command={"D_scheduled": 1.0, "frozen": False}, + semantics={"application": "epoch_boundary_after_all_recorded_ticks; affects subsequent epoch only", + "delta_dtype": "float32", "delta_units": "mV", "D_definition": "artificial scheduled scalar", + "actual_DAN_delivery_observed": False, "actual_DAN_delivery_used_to_gate": False}, + counts={"candidates": P.CANDIDATE_COUNT, "actual_bitwise_changed_edges": 1, "actual_value_changed_edges": 1}, + weights_installed=False, base_modified=False, brain_dynamics_executed=False, + scientific_learning_pass_claimed=False, candidate_generated=False) + rehash(receipt) + manifest = dict(schema="phase4an-overlay-v1", overlay_name="learned", overlay_sha256="1" * 64, + mask_path="/synthetic/mask.npy", mask_sha256=P.MASK_SHA256, + base_candidate_path="/synthetic/base.npy", base_candidate_sha256="2" * 64, + aj_model_dir="/synthetic/apl-timing-lab-v1", am_overlay_path="/synthetic/overlay.py", + am_overlay_sha256=P.AM_OVERLAY_SHA256, parent_snapshot_sha256="3" * 64, + eligibility_v2_source_sha256=P.PHASE4J_SHA256) + return before, after, after.copy(), receipt, manifest + + +class EpochPairTests(unittest.TestCase): + def setUp(self): + # Pure three-neuron fixtures; production constants remain full MaleCNS. + self.patches = [mock.patch.object(P, "N", 3), mock.patch.object(P, "E", 6), mock.patch.object(P, "N_KC", 2)] + for patch in self.patches: + patch.start() + self.addCleanup(patch.stop) + + def compare(self, reference, light, requests): + return P.compare_epoch_pair(reference, light, expected_requests=requests, + expected_steps=64, expected_effective_weight_sha256="e" * 64) + + def test_exact_actual_arrays_with_explicit_light_limits(self): + reference, light, requests = pair() + report = self.compare(reference, light, requests) + self.assertTrue(report["pass_all"], report["failed_checks"]) + self.assertTrue(report["numerical_exact"]) + self.assertTrue(report["nonvacuous_activity"]) + self.assertFalse(report["light_intermediate_states_observed"]) + self.assertFalse(report["light_per_tick_V_g_ref_identity_claimed"]) + self.assertEqual(report["recorded_pair_availability"], + {"APL_traces": False, "state_hashes": False, "queue_hashes": False}) + self.assertFalse(report["external_file_seals_process_and_input_authority_verified_here"]) + self.assertFalse(report["learning_pass_claimed"]) + self.assertFalse(report["candidate_generated"]) + json.dumps(report, allow_nan=False) + + def test_equal_quiet_records_are_not_new_activity_or_learning_pass(self): + reference, light, requests = pair(active=False) + report = self.compare(reference, light, requests) + self.assertTrue(report["numerical_exact"]) + self.assertFalse(report["nonvacuous_activity"]) + self.assertTrue(report["activity_is_descriptive_not_a_new_gate"]) + self.assertFalse(report["learning_pass_claimed"]) + + def test_signed_zero_final_voltage_difference_fails(self): + reference, light, requests = pair() + light["state"]["g"][0] = np.float64(0.) + report = self.compare(reference, light, requests) + self.assertFalse(report["pass_all"]) + self.assertIn("pair.state.g", report["failed_checks"]) + + def test_equal_but_wrong_original_requests_fail(self): + reference, light, requests = pair() + wrong = requests.copy() + wrong[0, 1] = 1 + report = self.compare(reference, light, wrong) + self.assertFalse(report["pass_all"]) + self.assertIn("reference.input_request_array", report["failed_checks"]) + self.assertIn("light.input_request_array", report["failed_checks"]) + + def test_reference_missing_diagnostics_cannot_be_excused_by_light(self): + for key in ("traces", "state_hashes", "queues"): + with self.subTest(key=key): + reference, light, requests = pair() + reference[key] = {} if key == "traces" else None + report = self.compare(reference, light, requests) + self.assertFalse(report["pass_all"]) + self.assertIn("reference.full_diagnostics", report["failed_checks"]) + + def test_partial_or_corrupt_array_schema_fails_closed(self): + reference, light, requests = pair() + del reference["traces"]["drive"] + with self.assertRaises(ValueError): + self.compare(reference, light, requests) + reference, light, requests = pair() + light["state"]["v"][0] = np.nan + with self.assertRaises(ValueError): + self.compare(reference, light, requests) + + def test_reported_modes_population_delay_and_scope_fail_closed(self): + mutations = (("nodes", 2), ("edges", 7), ("apl_delay_steps", 17), + ("diagnostic_mode", "reference"), ("first_failure_tick", 0), + ("within_trial_weight_updates", True), ("explicit_steps_executed", 63), + ("all_finite", 1), ("parent_checkpoint_manifest_sha256", "f" * 64)) + for key, value in mutations: + with self.subTest(key=key): + reference, light, requests = pair() + light["summary"][key] = value + self.assertFalse(self.compare(reference, light, requests)["pass_all"]) + + def test_effective_weight_and_base_coefficient_hash_binding(self): + for mutation in ("effective", "base_weights.npy", "coefficients.json"): + with self.subTest(mutation=mutation): + reference, light, requests = pair() + if mutation == "effective": + for record in (reference, light): + record["summary"]["effective_weights_raw_sha256"] = "f" * 64 + else: + light["summary"]["input_sha256"][mutation] = "f" * 64 + self.assertFalse(self.compare(reference, light, requests)["pass_all"]) + + def test_performance_is_not_compared_and_path_only_spec_identity_is_external(self): + reference, light, requests = pair() + reference["summary"].update(wall_seconds=1., peak_allocated_vram_bytes=1) + light["summary"].update(wall_seconds=99., peak_allocated_vram_bytes=999) + light["summary"]["input_sha256"]["spec.json"] = "f" * 64 + report = self.compare(reference, light, requests) + self.assertTrue(report["pass_all"]) + self.assertFalse(report["external_file_seals_process_and_input_authority_verified_here"]) + + def test_declared_light_full_diagnostics_or_derived_terminal_is_rejected(self): + for mutation in ("recorded", "origin"): + reference, light, requests = pair() + if mutation == "recorded": + light["summary"]["full_tick_state_hashes_recorded"] = True + else: + light["terminal_origin"] = "AM_spike_ring_derived_from_fresh_events_APL_bits_from_timing" + self.assertFalse(self.compare(reference, light, requests)["pass_all"]) + + def test_actual_summary_accounting_cannot_be_self_reported_wrong(self): + for key in ("requested", "accepted_requests", "rejected_requests", "total_spikes", "arriving_edges", "accepted_arriving_edges"): + with self.subTest(key=key): + reference, light, requests = pair() + light["summary"][key] += 1 + self.assertFalse(self.compare(reference, light, requests)["pass_all"]) + + def test_invalid_planned_request_seed_arrays_are_not_silently_coerced(self): + reference, light, requests = pair() + for bad in (requests.astype(np.int64), requests[::-1], np.concatenate((requests[:1], requests))): + with self.assertRaises(ValueError): + self.compare(reference, light, bad) + with self.assertRaises(ValueError): + P.compare_epoch_pair(reference, light, expected_requests=requests, expected_steps=True, + expected_effective_weight_sha256="e" * 64) + + +class DeltaBindingTests(unittest.TestCase): + def audit(self, values): + before, after, installed, receipt, manifest = values + return P.audit_delta_binding(before, after, installed, update_receipt=receipt, + parent_snapshot_sha256="3" * 64, child_manifest=manifest) + + def test_exact_array_binding_preserves_explicit_external_authority_limits(self): + values = delta_fixture() + for value in values[:3]: + value.setflags(write=False) + original = [value.tobytes() for value in values[:3]] + result = self.audit(values) + self.assertTrue(result["pass_all"], result["failed_checks"]) + self.assertEqual(result["actual_changed_edges"], 1) + for name in ("eligibility_v2_independently_recomputed_here", "full_mask_identity_verified_here", + "file_bytes_and_parent_snapshot_authority_verified_here", "installation_in_runtime_verified_here", + "actual_DAN_delivery_verified_here", "learning_pass_claimed"): + self.assertFalse(result[name]) + self.assertEqual([value.tobytes() for value in values[:3]], original) + json.dumps(result, allow_nan=False) + + def test_parent_overlay_file_sha_is_not_raw_delta_or_checkpoint_sha(self): + values = delta_fixture() + self.assertNotEqual(values[4]["overlay_sha256"], P._evidence(values[1])["raw_c_order_sha256"]) + self.assertTrue(self.audit(values)["pass_all"]) + for false_parent in (P._evidence(values[0])["raw_c_order_sha256"], "4" * 64): + values[4]["parent_snapshot_sha256"] = false_parent + result = self.audit(values) + self.assertFalse(result["pass_all"]) + self.assertIn("manifest.parent_overlay_file_sha", result["failed_checks"]) + + def test_installed_delta_signed_zero_and_row_mutation_rejected(self): + for mutation in ("zero", "row"): + values = delta_fixture() + if mutation == "zero": + values[2][0] = np.float32(-0.) + else: + values[2][1], values[2][2] = values[2][2], values[2][1] + result = self.audit(values) + self.assertFalse(result["pass_all"]) + self.assertIn("computed_installed_delta_bitwise", result["failed_checks"]) + + def test_rehashed_receipt_must_still_bind_supplied_arrays(self): + for location, key in (("input_arrays", "delta_before"), ("outputs", "delta_after")): + values = delta_fixture() + values[3][location][key]["raw_c_order_sha256"] = "f" * 64 + rehash(values[3]) + result = self.audit(values) + self.assertFalse(result["pass_all"]) + self.assertIn("receipt." + key, result["failed_checks"]) + + def test_rehashed_false_count_and_boolean_integer_ambiguity_rejected(self): + for field, value in (("actual_bitwise_changed_edges", 2), ("actual_value_changed_edges", True), ("candidates", float(P.CANDIDATE_COUNT))): + values = delta_fixture() + values[3]["counts"][field] = value + rehash(values[3]) + self.assertFalse(self.audit(values)["pass_all"]) + + def test_stale_receipt_or_changed_frozen_rule_manifest_rejected(self): + for mutation in ("hash", "v2", "mask", "AMhelper", "extra", "wrong_name"): + values = delta_fixture() + if mutation == "hash": + values[3]["payload_sha256"] = "0" * 64 + elif mutation == "extra": + values[4]["unregistered_override"] = True + elif mutation == "wrong_name": + values[4]["overlay_name"] = "static25" + else: + key = {"v2": "eligibility_v2_source_sha256", "mask": "mask_sha256", "AMhelper": "am_overlay_sha256"}[mutation] + values[4][key] = "0" * 64 + self.assertFalse(self.audit(values)["pass_all"]) + + def test_zero_D_and_frozen_require_exact_no_update_including_signed_zero(self): + for mode in ("D0", "frozen"): + values = delta_fixture() + values[3]["command"]["D_scheduled" if mode == "D0" else "frozen"] = 0.0 if mode == "D0" else True + rehash(values[3]) + result = self.audit(values) + self.assertFalse(result["pass_all"]) + self.assertIn("zero_D_or_frozen_identity", result["failed_checks"]) + values[1][:] = values[0] + values[2][:] = values[0] + values[3]["outputs"]["delta_after"] = P._evidence(values[1]) + values[3]["counts"].update(actual_value_changed_edges=0, actual_bitwise_changed_edges=0) + rehash(values[3]) + self.assertTrue(self.audit(values)["pass_all"]) + + def test_actual_DAN_or_immediate_US_update_claim_does_not_match_interface(self): + for field, value in (("actual_DAN_delivery_used_to_gate", True), ("actual_DAN_delivery_observed", True), + ("application", "at US tick"), ("delta_units", "V")): + values = delta_fixture() + values[3]["semantics"][field] = value + rehash(values[3]) + self.assertFalse(self.audit(values)["pass_all"]) + + def test_wrong_dtype_shape_nonfinite_or_positive_delta_rejected(self): + values = delta_fixture() + with self.assertRaises(ValueError): + P.audit_delta_binding(values[0].astype(np.float64), values[1], values[2], update_receipt=values[3], + parent_snapshot_sha256="3" * 64, child_manifest=values[4]) + values = list(delta_fixture()) + values[0] = values[0][:-1] + with self.assertRaises(ValueError): + self.audit(values) + values = delta_fixture() + values[0][0] = np.nan + with self.assertRaises(ValueError): + self.audit(values) + values = delta_fixture() + values[1][1] = values[2][1] = np.float32(.001) + values[3]["outputs"]["delta_after"] = P._evidence(values[1]) + rehash(values[3]) + result = self.audit(values) + self.assertFalse(result["pass_all"]) + self.assertIn("delta.nonpositive", result["failed_checks"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_inputs.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_inputs.py new file mode 100644 index 0000000..81072b1 --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_inputs.py @@ -0,0 +1,229 @@ +"""Pure/synthetic preflight tests: no actual rates, CUDA draws, or brain run.""" +import copy +import hashlib +import io +import json +from pathlib import Path +import tempfile +import types +import unittest +from unittest.mock import patch + +import numpy as np +import pandas as pd + +import conditioning_preflight_inputs as p + + +def synthetic_tables(): + names = ['synthetic-' + str(i) for i in range(12)] + names[6], names[7] = '4-ethylphenol', '3-hexanone' + ids = ['synthetic-id-' + str(i) for i in range(12)] + panel = pd.DataFrame(dict(odor_id=ids, odor_name=names, + split=['calibration'] * 6 + ['heldout'] * 6, mapped_response_sum=np.arange(1, 13, dtype=float), + formal_label=[''] * 6 + ['ODOR_A', 'ODOR_B'] + [''] * 4)) + selection = dict(selected_before_KC_simulation=True, glomeruli_order=['G2', 'G1'], + ODOR_A=dict(odor_id=ids[6], odor_name=names[6]), ODOR_B=dict(odor_id=ids[7], odor_name=names[7])) + vectors = {oid: np.array([.5, .25], np.float32) for oid in ids} + mapping = pd.DataFrame(dict(glomerulus=['G2', 'G2', 'G2', 'G1', 'G1'], + mapping_confidence=['exact', 'alias-normalized', 'exact', 'exact', 'unmatched'], + neuron_idx=[19., 3., 19., 8., 100.])) + config = dict(odor_encoder=dict(max_total_rate_per_glomerulus_hz=400., modes=['physiological', 'matched_energy'])) + return selection, panel, vectors, mapping, config + + +def sparse(rows): + values = np.asarray(rows, np.int32).reshape(-1, 2) + return {'tick': values[:, 0].copy(), 'neuron_idx': values[:, 1].copy()} + + +class PreflightTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.plan = p.build_preflight_plan() + cls.epoch = cls.plan['selected_epochs'][0] + source_parent = Path(__file__).resolve().parent.parent + candidates = (source_parent / 'phase4l.py', source_parent / 'models/male-cns-lif/phase4l.py') + path = next((candidate for candidate in candidates if candidate.is_file()), None) + if path is None: + raise FileNotFoundError('Pinned L source missing at both known server/local paths: ' + repr(candidates)) + cls.source_raw = path.read_bytes() + if hashlib.sha256(cls.source_raw).hexdigest() != p.SOURCE_PINS[p.SOURCE_REL]: + raise ValueError('Selected server/local L source differs from the frozen SHA') + cls.drive, cls.ast_sha = p._extract_drive(cls.source_raw, str(path)) + cls.tables, _ = p._rate_tables(*synthetic_tables(), cls.drive) + + def test_fixed_full_plan_master_first_two_real_epochs_and_12_calls(self): + full = self.plan['full_trial_plan'] + self.assertEqual(full['plan']['master'], 20260930) + self.assertEqual(full['payload_sha256'], 'a8e8620875e8213f3cc3b4f153fc5c06e1351bf1d91c7edaf4f18c1e2cdef62e') + self.assertEqual([e['epoch_index'] for e in self.plan['selected_epochs']], [0, 1]) + self.assertEqual([e['cs']['cue'] for e in self.plan['selected_epochs']], ['A', 'A']) + self.assertEqual([e['cs']['seed'] for e in self.plan['selected_epochs']], [5717173809506784725, 4342084263790487177]) + self.assertTrue(all(e['us']['requested'] for e in self.plan['selected_epochs'])) + self.assertEqual(self.plan['total_calls'], 12) + self.assertEqual(sum(r['steps'] for r in self.plan['calls']), 14400) + self.assertEqual(len({r['row_id'] for r in self.plan['calls']}), 12) + self.assertFalse(self.plan['complete_conditioning_authorized']) + + def test_plan_rows_share_inputs_but_keep_D_and_freeze_distinct(self): + for row in self.plan['calls']: + self.assertEqual(row['epoch_id'], self.plan['selected_epochs'][row['input_epoch_index']]['epoch_id']) + self.assertEqual(row['d_command'], 0. if row['arm'] == 'D0' else 1.) + self.assertEqual(row['frozen'], row['arm'] == 'frozen') + self.assertEqual(row['previous_delta_row_id'], None if row['input_epoch_index'] == 0 else + f"{row['arm']}--{row['diagnostic_mode']}--epoch-000") + self.assertEqual(self.plan['input_generation_calls'], 2) + + def test_whole_plan_validator_rejects_wrong_master_and_row_mutation(self): + # Reuse an already independently rebuilt expected fixture here; the + # production validator calls its full builder, not a count-only gate. + with patch.object(p, 'build_preflight_plan', return_value=self.plan) as rebuild: + self.assertTrue(p.validate_preflight_plan(copy.deepcopy(self.plan))) + bad = copy.deepcopy(self.plan); bad['master'] = 461 + with self.assertRaises(ValueError): p.validate_preflight_plan(bad) + bad = copy.deepcopy(self.plan); bad['calls'][0]['d_command'] = 0. + with self.assertRaises(ValueError): p.validate_preflight_plan(bad) + self.assertEqual(rebuild.call_count, 3) + + def test_actual_frozen_function_AST_preserves_mapping_order_rates_and_dtype(self): + table = self.tables['A'] + np.testing.assert_array_equal(table['targets'], [19, 3, 8]) + self.assertEqual(table['targets'].dtype, np.int32) + np.testing.assert_array_equal(table['rates_hz'], np.array([100, 100, 100], np.float32)) + self.assertEqual(table['source_panel_row'], 6) + self.assertEqual(table['source_panel_split'], 'heldout') # original static label only + self.assertFalse(table['targets'].flags.writeable) + self.assertFalse(table['rates_hz'].flags.writeable) + self.assertEqual(hashlib.sha256(self.source_raw).hexdigest(), p.SOURCE_PINS[p.SOURCE_REL]) + self.assertEqual(len(self.ast_sha), 64) + + def test_static_mapping_fails_closed_on_selection_order_schema_or_duplicate_PN(self): + for kind in ('name', 'split', 'vector', 'duplicate'): + args = list(synthetic_tables()) + if kind == 'name': args[0]['ODOR_A']['odor_name'] = 'different' + if kind == 'split': args[1].loc[0, 'split'] = 'heldout' + if kind == 'vector': args[2][args[0]['ODOR_A']['odor_id']] = np.array([np.nan, .25], np.float32) + if kind == 'duplicate': args[3].loc[3, 'neuron_idx'] = 19. + with self.assertRaises(ValueError): p._rate_tables(*args, type(self).drive) + + def test_loader_checks_all_static_bytes_then_uses_original_AST_no_old_import(self): + selection, panel, vectors, mapping, config = synthetic_tables() + buffer = io.BytesIO(); np.savez(buffer, glomeruli=np.array(selection['glomeruli_order']), **vectors) + payload = { + p.SOURCE_REL: self.source_raw, p.CONFIG_REL: json.dumps(config).encode(), + p.STATIC + 'odor-encoder/frozen-odor-panel.parquet': b'SYNTHETIC-PANEL', + p.STATIC + 'odor-encoder/mapped-glomerular-vectors.npz': buffer.getvalue(), + p.STATIC + 'odor-encoder/selection.json': json.dumps(selection).encode(), + p.STATIC + 'receptor-mapping/olfactory-mapping.parquet': b'SYNTHETIC-MAPPING', + } + pins = {k: hashlib.sha256(v).hexdigest() for k, v in payload.items()} + with tempfile.TemporaryDirectory() as td: + for name, raw in payload.items(): + target = Path(td) / name; target.parent.mkdir(parents=True, exist_ok=True); target.write_bytes(raw) + # Synthetic config is JSON (a YAML subset); local runtime lacks + # PyYAML. This mocks parsing only, not any scientific input/RNG. + yaml_mock = types.SimpleNamespace(safe_load=json.loads) + with patch.object(p, 'SOURCE_PINS', pins), patch.object(pd, 'read_parquet', side_effect=[panel, mapping]), \ + patch.dict('sys.modules', {'yaml': yaml_mock}): + result = p.load_rate_tables(td) + self.assertEqual(result['audit']['static_sources'], pins) + self.assertFalse(result['audit']['historical_dynamic_results_read']) + self.assertEqual(result['audit']['random_draws'], 0) + self.assertEqual(result['tables']['B']['odor_name'], '3-hexanone') + (Path(td) / p.CONFIG_REL).write_bytes(b'CHANGED') + with patch.object(p, 'SOURCE_PINS', pins), patch.object(pd, 'read_parquet') as parse, \ + patch.dict('sys.modules', {'yaml': yaml_mock}), self.assertRaises(ValueError): + p.load_rate_tables(td) + parse.assert_not_called() + + def test_generate_delegates_full_shape_full_seed_and_shared_US_without_real_draw(self): + fake = dict(requests=sparse([[2, 19], [999, 8]]), audit={'synthetic_sampler': True}) + dan = np.array([80, 90], np.int32) + with patch.object(p.inputs, 'generate_cs_requests', return_value=fake) as sampler: + result = p.generate_epoch_inputs(self.epoch, self.tables, dan, device='cuda:1', max_uniform_draws=3600) + args, kwargs = sampler.call_args + self.assertIs(args[0], self.tables['A']['targets']) + self.assertEqual(kwargs, dict(seed=5717173809506784725, steps=1200, cs_start_tick=0, cs_stop_tick=1000, + device='cuda:1', max_uniform_draws=3600)) + np.testing.assert_array_equal(result['requests']['tick'], [2, 900, 900, 999]) + np.testing.assert_array_equal(result['requests']['neuron_idx'], [19, 80, 90, 8]) + self.assertTrue(result['audit']['US_shared_across_arms']) + self.assertFalse(result['audit']['brain_reset_called']) + + def test_explicit_conflict_union_and_default_error_preserve_US_attempts(self): + fake = dict(requests=sparse([[900, 19]]), audit={'synthetic_sampler': True}) + with patch.object(p.inputs, 'generate_cs_requests', return_value=fake): + with self.assertRaises(p.inputs.RequestConflictError): + p.generate_epoch_inputs(self.epoch, self.tables, np.array([19], np.int32), device='cuda:0', max_uniform_draws=3600) + value = p.generate_epoch_inputs(self.epoch, self.tables, np.array([19], np.int32), + device='cuda:0', max_uniform_draws=3600, conflict_policy='boolean_union') + self.assertEqual(value['audit']['composition']['cross_source_conflict_count'], 1) + np.testing.assert_array_equal(value['source_mask'], [3]) + + def test_no_US_epoch_still_draws_CS_but_no_DAN_pulse(self): + epoch = copy.deepcopy(self.epoch) + epoch['us'].update(requested=False, tick=None, request_rule='none'); epoch['D_command'] = 0. + with patch.object(p.inputs, 'generate_cs_requests', return_value=dict(requests=sparse([]), audit={})): + value = p.generate_epoch_inputs(epoch, self.tables, np.array([80], np.int32), device='cuda:0', max_uniform_draws=3600) + self.assertEqual(len(value['us_requests']['tick']), 0) + + def test_invalid_targets_and_mutated_rates_rejected_before_sampling(self): + with patch.object(p.inputs, 'generate_cs_requests') as sampler: + for dan in (np.array([2, 2], np.int32), np.array([9, 2], np.int32), np.array([], np.int32), np.array([2], np.int64)): + with self.assertRaises(ValueError): + p.generate_epoch_inputs(self.epoch, self.tables, dan, device='cuda:0', max_uniform_draws=3600) + tables = copy.deepcopy(self.tables); tables['A']['rates_hz'][0] = 3. + with self.assertRaises(ValueError): + p.generate_epoch_inputs(self.epoch, tables, np.array([80], np.int32), device='cuda:0', max_uniform_draws=3600) + sampler.assert_not_called() + + def original_spec(self): + return dict(name='old-case', kind='fullgraph', steps=1000, dt_ms=.1, duration_ms=100., + model={'immutable': 'synthetic model'}, graph_dir='/old/graph', apl_bundle_dir='/old/apl', + coefficients_path='/old/coeff.json', science_python='/old/venv/bin/python', + apl_delay_steps=18, precision='float64', apl_precision='float32', record_history=False, + training=False, validation_cohort='old_cohort', odor_index=0, expected_external_events=123., + odor_id='old-id', odor_name='old-name', source_panel_row=0, source_panel_split='calibration', + rate_sum_hz=999., target_count=100, seed=22, budget_root='/old/experiment', output_budget_bytes=100) + + def test_derive_spec_changes_only_explicit_input_scope_and_clock_budget_fields(self): + old = self.original_spec(); before = copy.deepcopy(old) + value, audit = p.derive_case_spec(old, self.plan['calls'][0], self.epoch, self.tables['A'], + brain_reset_seed=20260930, budget_root='/new/preflight', output_budget_bytes=10000) + self.assertEqual(old, before) + self.assertEqual((value['steps'], value['duration_ms'], value['max_steps']), (1200, 120., 1200)) + self.assertEqual(value['seed'], 20260930) + self.assertEqual(value['cs_input_seed'], 5717173809506784725) + for key in ('model', 'graph_dir', 'apl_bundle_dir', 'coefficients_path', 'science_python'): + self.assertEqual(value[key], old[key]) + self.assertFalse(value['within_trial_weight_updates']) + self.assertEqual(value['odor_name'], '4-ethylphenol') + self.assertNotIn('validation_cohort', value); self.assertNotIn('training', value) + self.assertEqual(set(audit['removed_stale_input_metadata']), {'validation_cohort', 'training', 'odor_index', 'expected_external_events'}) + self.assertFalse(audit['full_conditioning_authorized']) + json.dumps(audit, allow_nan=False) + + def test_seed_or_neural_continuation_or_budget_alias_invalid(self): + args = dict(brain_reset_seed=3, budget_root='/new/preflight', output_budget_bytes=10000) + for seed in (-1, 2**31, self.epoch['cs']['seed'], True, np.int32(3)): + with self.assertRaises(ValueError): + p.derive_case_spec(self.original_spec(), self.plan['calls'][0], self.epoch, self.tables['A'], **{**args, 'brain_reset_seed': seed}) + for root in ('/old/experiment/new', '/old', '/new/../old', 'relative'): + with self.assertRaises(ValueError): + p.derive_case_spec(self.original_spec(), self.plan['calls'][0], self.epoch, self.tables['A'], **{**args, 'budget_root': root}) + old = self.original_spec(); old['parent_checkpoint_manifest_sha256'] = 'a' * 64 + with self.assertRaises(ValueError): p.derive_case_spec(old, self.plan['calls'][0], self.epoch, self.tables['A'], **args) + + def test_row_command_mismatch_and_rate_label_mismatch_fail_closed(self): + row = copy.deepcopy(self.plan['calls'][0]); row['d_command'] = 0. + with self.assertRaises(ValueError): + p.derive_case_spec(self.original_spec(), row, self.epoch, self.tables['A'], + brain_reset_seed=3, budget_root='/new/preflight', output_budget_bytes=10000) + with self.assertRaises(ValueError): + p.derive_case_spec(self.original_spec(), self.plan['calls'][0], self.epoch, self.tables['B'], + brain_reset_seed=3, budget_root='/new/preflight', output_budget_bytes=10000) + + +if __name__ == '__main__': + unittest.main() diff --git a/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_prepare.py b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_prepare.py new file mode 100644 index 0000000..19c03a1 --- /dev/null +++ b/models/male-cns-lif/olfactory-conditioning-lab-v1/test_conditioning_preflight_prepare.py @@ -0,0 +1,109 @@ +"""CPU-only coordinator contract tests; no real brain/input sampling.""" +import ast +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +import numpy as np + +import conditioning_preflight_prepare as P +import conditioning_preflight_workflow as W + + +class CoordinatorTests(unittest.TestCase): + def test_npz_exclusive_roundtrip(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / 'x.npz' + a = np.asarray([0, 3], np.int32) + P.save_npz(path, tick=a) + self.assertEqual(W.npz(path)['tick'].tobytes(), a.tobytes()) + with self.assertRaises(FileExistsError): P.save_npz(path, tick=a) + + def test_exact_pins_and_mutation(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / 'input' + p.write_bytes(b'original') + pins = {str(p): P.digest(p)} + P.verify_pins(pins) + p.write_bytes(b'changed') + with self.assertRaisesRegex(ValueError, 'Pinned file'): P.verify_pins(pins) + + def test_rows_exact_unique(self): + row = dict(row_id='D1--reference--epoch-000') + self.assertEqual(W.row_by({'plan': {'calls': [row]}}, row['row_id']), row) + for rows in ([], [row, row]): + with self.assertRaises(ValueError): W.row_by({'plan': {'calls': rows}}, row['row_id']) + + def test_locations_reject_path_traversal(self): + for value in ('../x', '/x', 'a/b', 'a\\b', '', None): + with self.assertRaises(ValueError): W.locations(value) + + def test_plan_descriptor_avoids_huge_payload_duplication(self): + value = dict(full_trial_plan={'large': [1]}, payload_sha256='x', calls=[{'row_id': 'a'}]) + self.assertEqual(P.plan_descriptor(value), {'calls': [{'row_id': 'a'}]}) + self.assertIn('full_trial_plan', value) + + def test_original_AST_rejects_unanchored_bytes(self): + with self.assertRaisesRegex(ValueError, 'L source changed'): + W.original_sampler_nodes(b'def run_rates(): pass') + + def test_original_AST_if_local_frozen_source_available(self): + # Both repository checkout and local source mirror are supported; this + # test never reads historical dynamics or executes the extracted code. + candidates = [Path(__file__).resolve().parents[1] / 'phase4l.py', + Path(__file__).resolve().parents[1] / 'models/male-cns-lif/phase4l.py'] + source = next((p for p in candidates if p.is_file()), None) + self.assertIsNotNone(source, 'Original L source required for actual AST extraction test') + nodes = W.original_sampler_nodes(source.read_bytes()) + self.assertEqual(len(nodes), 4) + self.assertNotIn('reset', ast.unparse(ast.Module(body=nodes, type_ignores=[]))) + + def test_registration_external_hash_rejection_before_parse(self): + with tempfile.TemporaryDirectory() as d: + exp = Path(d); (exp / 'protocol').mkdir() + P.save(exp / 'protocol/preregistered.json', {'fake': True}) + with patch.object(P, 'EXP', exp): + with self.assertRaisesRegex(ValueError, 'external anchor'): + P.read_registration('0' * 64) + + def test_initial_overlay_has_no_parent_process_lookup(self): + expected = dict(overlay_file='/immutable/zero.npz', overlay_manifest='/immutable/zero.json') + self.assertIs(W.before_overlay({'initial_overlay': expected}, 'unused', {'previous_delta_row_id': None}), expected) + + def test_wrong_parent_arm_rejected_before_runtime_read(self): + parent = dict(row_id='parent', arm='D0', diagnostic_mode='light', input_epoch_index=0) + child = dict(previous_delta_row_id='parent', arm='D1', diagnostic_mode='light', input_epoch_index=1) + with self.assertRaisesRegex(ValueError, 'cross-epoch'): + W.before_overlay({'plan': {'calls': [parent]}}, 'unused', child) + + def test_budget_refuses_overrun_without_delete(self): + with tempfile.TemporaryDirectory() as d: + exp = Path(d); p = exp / 'keep' + p.write_bytes(b'1234') + with patch.object(W, 'EXP', exp): + with self.assertRaisesRegex(ValueError, 'Resource cap'): + W.budget({'budget': {'output_budget_bytes': 1, 'minimum_free_reserve_bytes': 1}}) + self.assertEqual(p.read_bytes(), b'1234') + + def test_partial_log_is_not_retried(self): + with tempfile.TemporaryDirectory() as d: + exp = Path(d); (exp / 'logs').mkdir() + (exp / 'logs/input-generation.stdout.log').write_bytes(b'partial') + with patch.object(W, 'EXP', exp), patch.object(W.subprocess, 'run') as run: + with self.assertRaisesRegex(ValueError, 'never retry'): W.child_process({}, 'sha', 'materialize') + run.assert_not_called() + + def test_admission_blocks_stopped_complete_or_sealed(self): + with tempfile.TemporaryDirectory() as d: + exp = Path(d); (exp / 'audit').mkdir() + with patch.object(W, 'EXP', exp): + W.admission() + for name in ('audit/stopped.json', 'audit/execution-complete.json', 'manifest.json'): + p = exp / name; p.write_bytes(b'{}') + with self.assertRaisesRegex(ValueError, 'Stopped/completed'): W.admission() + p.unlink() + + +if __name__ == '__main__': + unittest.main()