From 8abfa3c66cbccbfef885cc30b926dbcbdd3cca8e Mon Sep 17 00:00:00 2001 From: Digital Drosophila Archive Date: Thu, 1 Oct 2026 14:30:01 +0800 Subject: [PATCH] Add isolated exact eligibility GPU replay prototype and fixed benchmark protocol --- .../conditioning-acceleration-v1/PROTOCOL.md | 25 + .../conditioning-acceleration-v1/benchmark.py | 216 ++++++++ .../export_fixtures.py | 518 ++++++++++++++++++ .../export_synthetic.py | 242 ++++++++ .../gpu_backend.py | 458 ++++++++++++++++ .../repair_environment.py | 22 + .../requirements-candidate.txt | 2 + .../setup_environment.py | 33 ++ .../test_export_fixtures.py | 135 +++++ .../test_gpu_math.py | 233 ++++++++ 10 files changed, 1884 insertions(+) create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/PROTOCOL.md create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/benchmark.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/export_fixtures.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/gpu_backend.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/repair_environment.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/requirements-candidate.txt create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/setup_environment.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/test_export_fixtures.py create mode 100644 models/male-cns-lif/conditioning-acceleration-v1/test_gpu_math.py diff --git a/models/male-cns-lif/conditioning-acceleration-v1/PROTOCOL.md b/models/male-cns-lif/conditioning-acceleration-v1/PROTOCOL.md new file mode 100644 index 0000000..1882606 --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/PROTOCOL.md @@ -0,0 +1,25 @@ +# Conditioning acceleration v1 — independent engineering benchmark + +Authority: user requested V100 acceleration and algorithm optimization on 2026-10-01. This work DOES NOT resume or change Phase 4AR, and does not authorize a new 30,000-epoch scientific experiment. + +## Frozen scientific semantics + +Preserve original Phase 4J/AM eligibility, compartment mapping, the complete candidate mask, anatomical base weights, float32 update semantics, all time steps and original event ordering. No threshold/gain/eta/tau/sign change, reduced trial count claim, surrogate gradients, skip-time power approximation, new connections or model learning claim. GPU outputs are diagnostic replay products only and are never installed into a brain. + +The oracle is the exact SHA-pinned original Torch CPU update running in the original science environment. Candidate CuPy CUDA code runs in a separate environment with NumPy 2.2.6 and CuPy 13.6.0, using the installed CUDA 12.6 toolkit. Do not modify the original environment, ComfyUI or frozen source files. + +## Fixed validation and measurement plan + +Select the first and last fully ledger-committed training task for each observed arm by deterministic task-ID order, without inspecting learning scores. This is coverage of available saved inputs only, not all registered conditions. Export fixed original inputs and recomputed CPU reference outputs into a new directory; compare them with the saved scientific outputs before use. + +Compare all sampled traces, final traces and local delta arrays byte-for-byte against the original CPU oracle on both physical V100s and the 4090. Add synthetic edge cases: no spikes, non-KC spikes, no US, US at zero/final tick, duplicate/invalid events rejection, multiple US snapshots for trace-only API, saturation, D=0, frozen identity, all compartments, and seeded sparse/random traces. Do not select only passing cases. Any mismatch blocks promotion. + +Use 3 warmups and 20 timed repeats for every real fixture on each GPU, with the same fixture order. Record first-use compilation/context cost separately, CPU preparation, transfer-inclusive host wall time, kernel event time, GPU identity, memory and exact software/source hashes. Repeat within a resident process to expose avoidable per-process overhead. Never present kernel-only speedup as full training speedup. + +Budget: at most 15 minutes of candidate benchmark execution after dependencies/compilation, one process per GPU initially; do not occupy a GPU with a nonempty ComfyUI queue or evict services. No automatic scientific retry. Implementation failures may be diagnosed and corrected in new benchmark attempts, retaining earlier reports. No modification of any frozen experiment. Independent CPU verification remains required before any future production integration. + +## Gate and limits + +Passing this benchmark means only ELIGIBILITY-UPDATE-REPLAY-EQUIVALENCE for the tested inputs. It does not establish full-brain event equivalence, end-to-end acceleration, learning success, resumability, or completion of Phase 4AR. Integration, persistent brain workers and independent-seed multi-GPU scheduling require separately validated execution/audit changes. Stop before installing any update or starting learning. + +Keep source, configuration, environment installation report, fixed input manifest, raw timing samples, comparisons, failures and SHA-256 manifest. Check Phase 4AR registration/source and the prior saved-file inventory before/after; report the exact scope (not a new Phase 1–4AQ numerical regression). diff --git a/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py b/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py new file mode 100644 index 0000000..c6be031 --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py @@ -0,0 +1,216 @@ +"""Bounded replay benchmark. No brain simulation, training, or weight installation.""" +from __future__ import annotations +import argparse +from datetime import datetime, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import statistics +import subprocess +import sys +import time +import traceback +import urllib.request + +import numpy as np +from gpu_backend import CudaEligibility, prepare_events + +REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740' +IDENTITY_KEYS = ('kc', 'candidate_edge_id', 'candidate_sources', 'candidate_targets', + 'candidate_compartments', 'base_candidate') +OUTPUTS = ('sampled_trace', 'final_trace', 'snapshots', 'delta_after') + +def sha(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 save(path, value): + with Path(path).open('x', encoding='utf-8') as stream: + json.dump(value, stream, indent=2, sort_keys=True, allow_nan=False) + stream.write('\n') + +def read(path): + return json.loads(Path(path).read_text()) + +def evidence(a): + return dict(dtype=a.dtype.str, shape=list(a.shape), sha256=hashlib.sha256(a.tobytes()).hexdigest()) + +def contained(root, relative): + p = root / relative + assert not p.is_symlink() and p.resolve().is_relative_to(root.resolve()), 'Escaping fixture path' + return p + +def idle_queues(): + result = {} + for port in (8188, 8189, 8190): + with urllib.request.urlopen(f'http://192.168.10.51:{port}/queue', timeout=5) as response: + q = json.load(response) + assert set(('queue_running', 'queue_pending')).issubset(q) + result[str(port)] = {key: len(q[key]) for key in ('queue_running', 'queue_pending')} + assert not any(result[str(port)].values()), 'ComfyUI has work; benchmark will not compete' + return result + +def gpu_snapshot(uuid): + text = subprocess.check_output(['nvidia-smi', '--id=' + uuid, + '--query-gpu=uuid,name,pci.bus_id,memory.total,memory.used,utilization.gpu,driver_version', + '--format=csv,noheader,nounits'], text=True).strip() + fields = [s.strip() for s in text.split(',')] + assert fields[0] == uuid + return dict(zip(('uuid', 'name', 'pci_bus_id', 'memory_total_mib', 'memory_used_mib', + 'utilization_percent', 'driver_version'), fields)) + +def stats(values): + return dict(n=len(values), min=min(values), median=statistics.median(values), + mean=statistics.mean(values), max=max(values)) + +def compare(result, arrays): + report = {} + for key in OUTPUTS: + actual, expected = getattr(result, key), arrays['reference_' + key] + exact = actual.dtype == expected.dtype and actual.shape == expected.shape and actual.tobytes() == expected.tobytes() + diff = np.abs(actual.astype(np.float64) - expected.astype(np.float64)) + report[key] = dict(bitwise_exact=exact, actual=evidence(actual), expected=evidence(expected), + max_abs_error=float(np.max(diff)) if diff.size else 0., + mean_abs_error=float(np.mean(diff)) if diff.size else 0.) + return report + +def load_fixtures(manifests): + fixtures = [] + common_identity = None + for manifest_path in manifests: + m = read(manifest_path) + assert m.get('complete') is True and m['registration_sha256'] == REG_SHA + root = manifest_path.parent + for f in m['fixtures']: + npz = contained(root, f['npz']) + command_path = contained(root, f['command']) + assert sha(npz) == f['npz_sha256'] and sha(command_path) == f['command_sha256'] + with np.load(npz, allow_pickle=False) as z: + a = {k: z[k].copy() for k in z.files} + identity = {k: evidence(a[k]) for k in IDENTITY_KEYS} + if common_identity is None: + common_identity = identity + assert common_identity == identity, 'Full KC/mask/base identity differs across fixtures' + assert len(a['kc']) == 4064 and len(a['candidate_edge_id']) == 14232 + source_pins = read(npz.parent / 'reference-sources.json') + for info in source_pins.values(): + assert sha(info['path']) == info['sha256'], 'Reference source changed' + # Extract constants from the authenticated original bridge, without import/exec. + import ast + bridge = source_pins['eligibility_bridge_source'] + assert bridge['sha256'] == '174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61' + constants = {} + for node in ast.parse(Path(bridge['path']).read_text()).body: + if isinstance(node, ast.Assign): + for target in node.targets: + if isinstance(target, ast.Name) and target.id in ('DT_MS', 'TAU_MS', 'ETA_MV', 'FLOOR_FRACTION', 'POLARITY'): + constants[target.id] = ast.literal_eval(node.value) + assert constants['POLARITY'] == -1 + rule = dict(eta_mv=constants['ETA_MV'], floor_fraction=constants['FLOOR_FRACTION'], + decay=math.exp(-constants['DT_MS'] / constants['TAU_MS'])) + fixtures.append(dict(group=root.name, record=f, arrays=a, command=read(command_path), + rule=rule, input_path=str(npz), manifest_sha256=sha(manifest_path))) + assert fixtures + return fixtures, common_identity + +def main(args): + assert os.environ.get('CUDA_VISIBLE_DEVICES') == args.uuid, 'Use one explicit physical UUID per process' + assert args.output.is_absolute() and not args.output.exists() + args.output.mkdir() + started = time.perf_counter() + try: + queues_before = idle_queues() + gpu_before = gpu_snapshot(args.uuid) + fixtures, common_identity = load_fixtures(args.manifest) + backend = CudaEligibility(device_index=0) + cp = backend.cp + props = cp.cuda.runtime.getDeviceProperties(0) + bus = cp.cuda.runtime.deviceGetPCIBusId(0) + if isinstance(bus, bytes): bus = bus.decode() + assert bus.lower()[-10:] == gpu_before['pci_bus_id'].lower()[-10:], 'Physical GPU binding differs' + device = dict(**gpu_before, runtime_pci_bus_id=bus, compute_capability=backend.device.compute_capability, + initialization_seconds=backend.initialization_seconds, cupy_version=cp.__version__, + numpy_version=np.__version__, python_version=sys.version, + runtime_version=cp.cuda.runtime.runtimeGetVersion(), driver_version_runtime=cp.cuda.runtime.driverGetVersion(), + runtime_name=props['name'].decode() if isinstance(props['name'], bytes) else props['name']) + source = Path(__file__).resolve().parent + commit = subprocess.check_output(['git', '-C', str(source), 'rev-parse', 'HEAD'], text=True).strip() + assert subprocess.check_output(['git', '-C', str(source), 'status', '--porcelain'], text=True) == '', 'Candidate source worktree dirty' + save(args.output / 'authority.json', dict(device=device, queues_before=queues_before, + candidate_commit=commit, common_full_mapping_identity=common_identity, + fixture_manifests={str(p): sha(p) for p in args.manifest}, repeats=20, warmups=3, + all_source_files={p.name: sha(p) for p in sorted(source.iterdir()) if p.is_file()}, + scope='eligibility/delta replay only; no brain, training, or installation', + original_experiment_resumed=False, created_utc=datetime.now(timezone.utc).isoformat())) + reports = [] + peak_reserved = 0 + for i, fixture in enumerate(fixtures): + assert time.perf_counter() - started < 900, 'Benchmark wall budget exhausted' + idle_queues() + a, c, rule = fixture['arrays'], fixture['command'], fixture['rule'] + arrays_before = {k: evidence(v) for k, v in a.items()} + f_id = fixture['group'] + '--' + fixture['record']['fixture_id'] + def run_once(resident=False): + begin = time.perf_counter() + prepared = prepare_events(a['step'], a['neuron_idx'], a['kc'], c['steps'], + a['reference_command_steps'], n_neurons=c['n_neurons']) + result = backend.run(prepared, a['candidate_sources'], a['candidate_compartments'], + c['compartment'], a['delta_before'], a['base_candidate'], + d=c['d_command'], frozen=c['frozen'], **rule, + kernel_warmup=3 if resident else 0, kernel_repeats=20 if resident else 1) + comparison = compare(result, a) + elapsed = time.perf_counter() - begin + if not all(r['bitwise_exact'] for r in comparison.values()): + save(args.output / (f_id + '-mismatch.json'), comparison) + raise AssertionError('GPU bitwise mismatch: ' + f_id) + return result, comparison, elapsed + cold, comp, cold_wall = run_once() + for _ in range(3): + run_once() + samples = [] + for _ in range(20): + result, comp, elapsed = run_once() + samples.append(dict(**result.timing, replay_wall_including_preparation_and_comparison_seconds=elapsed)) + resident, _, resident_wall = run_once(resident=True) + peak_reserved = max(peak_reserved, cp.get_default_memory_pool().total_bytes()) + assert arrays_before == {k: evidence(v) for k, v in a.items()}, 'Input arrays mutated' + report = dict(fixture_id=f_id, input_path=fixture['input_path'], fixture_manifest_sha256=fixture['manifest_sha256'], + command=c, rule=rule, first_call_timing=cold.timing, first_call_wall_seconds=cold_wall, + raw_warm_samples=samples, resident_timing=resident.timing, + summaries={key: stats([s[key] for s in samples]) for key in + ('preparation_seconds', 'kernel_seconds', 'end_to_end_seconds', + 'replay_wall_including_preparation_and_comparison_seconds')}, + comparison=comp, input_unchanged=True, all_repeats_bitwise_exact=True, + reference_timing=fixture['record'].get('reference_timing'), audit=result.audit) + save(args.output / (f_id + '.json'), report) + reports.append(report) + print(json.dumps(dict(fixture=f_id, exact=True, + median_ms=1000 * report['summaries']['replay_wall_including_preparation_and_comparison_seconds']['median'])), flush=True) + # Recheck original exported files at the end, not just at read time. + load_fixtures(args.manifest) + result = dict(status='PASS-ELIGIBILITY-UPDATE-REPLAY-EQUIVALENCE', fixtures=len(reports), + device=device, candidate_commit=commit, all_fixture_repeats_bitwise_exact=True, + observed_cuda_pool_reserved_peak_bytes=peak_reserved, gpu_after=gpu_snapshot(args.uuid), + queues_after=idle_queues(), wall_seconds=time.perf_counter()-started, + original_experiment_resumed=False, brain_simulation_tested=False, + full_training_speedup_measured=False, learning_success_claimed=False, + coverage_limit='available prefix + fixed synthetic edge cases, not all registered conditions', + files={p.name: sha(p) for p in sorted(args.output.iterdir()) if p.is_file()}) + save(args.output / 'summary.json', result) + print(json.dumps(result), flush=True) + except Exception: + save(args.output / 'failure.json', dict(exit_code=1, traceback=traceback.format_exc(), + wall_seconds=time.perf_counter()-started, original_experiment_resumed=False)) + raise + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--manifest', type=Path, action='append', required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--uuid', required=True) + main(parser.parse_args()) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/export_fixtures.py b/models/male-cns-lif/conditioning-acceleration-v1/export_fixtures.py new file mode 100644 index 0000000..12212b9 --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/export_fixtures.py @@ -0,0 +1,518 @@ +"""Export CPU reference fixtures from the fixed, committed Phase 4AR prefix. + +Run with the ORIGINAL models/male-cns-lif/.venv/bin/python -B and --output an +absolute, nonexistent directory outside frozen input/source directories. This +tool never launches a brain, resumes training, installs weights or changes AR. +The original AQ CPU validator is run in fresh CPU-only subprocesses; its only +new receipts are placed inside the requested output. A second fresh process per +fixture computes the original U Torch CPU reference, including a warm repeat. + +Selection is first and last task_id in each available arm among committed +brain/update/verify triples, determined before result/metric files are opened. +This bounded engineering sample does not cover all scientific conditions. +""" +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import time + + +ROOT = Path('/data/digital-drosophila') +MODEL = ROOT / 'models/male-cns-lif' +EXP = ROOT / 'experiments/phase4ar-normal-conditioning' +ORIGINAL_ENV = MODEL / '.venv' +REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740' +SOURCE_COMMIT = '80aeb2093db655329774eff7b32fb6e417b8b40b' +U_SHA = '96efff678cff910654eb08e86f1a17789e4e5589e340345f419c7027a7d547e2' +BRIDGE_SHA = '174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61' +J_SHA = 'abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e' +ARMS = ('cs_only', 'd_zero', 'dan_only', 'frozen', 'paired', 'random_reward', 'unpaired') +INPUT_ARRAYS = ('step', 'neuron_idx', 'kc', 'candidate_edge_id', 'candidate_sources', + 'candidate_targets', 'candidate_compartments', 'delta_before', 'base_candidate') +OUTPUT_ARRAYS = ('delta_after', 'sampled_trace', 'final_trace') + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False) + + +def digest(path): + h = hashlib.sha256() + with Path(path).open('rb') as stream: + for chunk in iter(lambda: stream.read(8 << 20), b''): + h.update(chunk) + return h.hexdigest() + + +def unlinked(path): + path = Path(path) + require(path.is_absolute() and '..' not in path.parts, 'Absolute nonescaping path required') + require(not any(p.is_symlink() or (hasattr(p, 'is_junction') and p.is_junction()) + for p in (path, *path.parents)), 'Linked path forbidden: ' + str(path)) + return path + + +def read(path): + def pairs(items): + result = {} + for key, value in items: + require(key not in result, 'Duplicate JSON key') + result[key] = value + return result + return json.loads(unlinked(path).read_text(encoding='utf-8'), object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(ValueError('Nonfinite JSON'))) + + +def save(path, value): + with unlinked(path).open('x', encoding='utf-8', newline='\n') as stream: + json.dump(value, stream, sort_keys=True, indent=2, ensure_ascii=False, allow_nan=False) + stream.write('\n') + + +def identity(path): + path = unlinked(path) + require(path.is_file(), 'Missing regular file: ' + str(path)) + size = path.stat().st_size + result = dict(sha256=digest(path), bytes=size) + require(path.stat().st_size == size, 'File size changed while hashing') + return result + + +def check(path, expected): + observed = identity(path) + sha = expected if isinstance(expected, str) else expected['sha256'] + require(observed['sha256'] == sha, 'Pinned bytes changed: ' + str(path)) + if isinstance(expected, dict) and 'bytes' in expected: + require(observed['bytes'] == expected['bytes'], 'Pinned size changed: ' + str(path)) + return observed + + +def task_key(task): + return 'task-' + hashlib.sha256(task['task_id'].encode()).hexdigest() + + +def select_tasks(tasks, committed_ids): + """Only plan metadata and committed entry IDs participate in selection.""" + available = {arm: [] for arm in ARMS} + seen = set() + for task in tasks: + require(task['task_id'] not in seen, 'Duplicate plan task identity') + seen.add(task['task_id']) + if task['kind'] != 'train': + continue + require(task['arm'] in available, 'Unexpected training arm') + key = task_key(task) + if all(key + '--' + stage in committed_ids for stage in ('brain', 'update', 'verify')): + available[task['arm']].append(task) + selected = [] + for arm in ARMS: + ordered = sorted(available[arm], key=lambda task: task['task_id']) + if ordered: + selected.append(ordered[0]) + if ordered[-1]['task_id'] != ordered[0]['task_id']: + selected.append(ordered[-1]) + require(selected, 'No committed train triple is available') + return selected, {arm: len(available[arm]) for arm in ARMS} + + +def ledger_snapshot(exp, reg): + """Read ledger controls, not all historical outcomes; never instantiate Ledger.""" + control = exp / '.append-ledger' + require(not (control / 'writer.lock').exists(), 'Ledger writer/stale lock present') + header_path = control / 'header.json' + header = read(header_path) + require(header['schema'] == 'normal-append-ledger-v1' and header['root'] == str(exp) + and header['registration_sha256'] == REG_SHA, 'Ledger header authority differs') + expected_initial = {str(Path(p).relative_to(exp)): sha for p, sha in reg['prepared_sha256'].items()} + expected_initial['protocol/preregistered.json'] = REG_SHA + require({p: info['sha256'] for p, info in header['initial_inventory'].items()} == expected_initial, + 'Ledger initial inventory differs from registration') + control_pins = {str(header_path): identity(header_path)} + tracked = {str(exp / p): info for p, info in header['initial_inventory'].items()} + entries = {} + tail = None + paths = sorted((control / 'entries').iterdir()) + for sequence, path in enumerate(paths, 1): + require(path.name == f'{sequence:016d}.json', 'Ledger entries must form a contiguous prefix') + row = read(path) + name = row['entry_id'] + require(name not in entries and row['schema'] == 'normal-append-ledger-v1-entry' + and row['registration_sha256'] == REG_SHA and row['sequence'] == sequence + and row['prev_entry_sha256'] == tail and row['scientific_gate_claimed'] is False, + 'Ledger chain identity differs') + start_path = control / 'started' / (name + '.json') + start = read(start_path) + control_pins[str(start_path)] = check(start_path, row['start_sha256']) + require(start['schema'] == 'normal-append-ledger-v1-start' and start['entry_id'] == name + and start['registration_sha256'] == REG_SHA and start['next_sequence'] == sequence + and start['prev_entry_sha256'] == tail and start['retry_authorized'] is False, + 'Ledger start identity differs') + for output in row['new_outputs']: + output_path = unlinked(output['path']) + require(output_path.is_relative_to(exp) and str(output_path.relative_to(exp)) == output['relative_path'] + and str(output_path) not in tracked, 'Output escapes or repeats immutable history') + tracked[str(output_path)] = {key: output[key] for key in ('bytes', 'sha256')} + control_pins[str(path)] = identity(path) + tail = control_pins[str(path)]['sha256'] + entries[name] = dict(row=row, start=start, ledger_path=str(path), ledger_sha256=tail) + require(entries, 'Empty committed prefix') + return entries, tracked, control_pins, dict(entries=len(entries), head_sha256=tail, + uncommitted_starts=sorted(p.stem for p in (control / 'started').glob('*.json') if p.stem not in entries), + scope='ledger chain controls and selected artifacts; not a full historical payload audit') + + +def validate_process(entry, tracked, exp): + row, start = entry['row'], entry['start'] + name = row['entry_id'] + metadata = row['metadata'] + job_path = unlinked(metadata['job_file']) + process_path = unlinked(metadata['process_receipt_file']) + require(job_path == exp / 'jobs' / (name + '.json') and process_path == exp / 'receipts' / name / 'process.json', + 'Process/job location differs') + require(str(job_path) in tracked and str(process_path) in tracked, 'Process/job not in committed history') + job_pin = check(job_path, tracked[str(job_path)]) + process_pin = check(process_path, tracked[str(process_path)]) + require(start['job']['path'] == str(job_path) and start['job']['sha256'] == job_pin['sha256'], + 'Started job differs from committed job') + process = read(process_path) + require(process['entry_id'] == name and process['registration_sha256'] == REG_SHA + and process['job_sha256'] == job_pin['sha256'] and process['complete'] is True + and type(process['exit_code']) is int and process['exit_code'] == 0 + and process['timed_out'] is False and process['exit_observed_by_parent'] is True + and process.get('launch_error') is None and type(metadata['observed_returncode']) is int + and metadata['observed_returncode'] == 0, 'Selected process is not committed observed exit zero') + require(process['argv'][0] == str(ORIGINAL_ENV / 'bin/python') and '-B' in process['argv'], + 'Selected process did not use original -B interpreter') + job = read(job_path) + require(job['registration_sha256'] == REG_SHA, 'Selected job registration differs') + outgoing = {item['path']: item for item in row['new_outputs']} + require(str(job_path) in outgoing and str(process_path) in outgoing, + 'Job/process must be outputs of this exact committed entry') + consumed = {item['path']: item['sha256'] for item in row['consumed_inputs']} + require(all(consumed.get(path) == sha for path, sha in job['input_sha256'].items()), + 'Selected job input pins differ from committed ledger') + return job, {str(job_path): job_pin, str(process_path): process_pin}, process + + +def array_evidence(array): + return dict(dtype=array.dtype.str, shape=list(array.shape), nbytes=int(array.nbytes), + raw_c_order_sha256=hashlib.sha256(array.tobytes(order='C')).hexdigest()) + + +def load_npz(path): + import numpy as np + with np.load(unlinked(path), allow_pickle=False) as archive: + return {name: archive[name].copy() for name in archive.files} + + +def save_npz(path, arrays): + import numpy as np + with unlinked(path).open('xb') as stream: + np.savez_compressed(stream, **arrays) + + +def load_verified_source(name, path, sha): + raw = unlinked(path).read_bytes() + require(hashlib.sha256(raw).hexdigest() == sha and name not in sys.modules, 'Unverified source/module binding') + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + exec(compile(raw, str(path), 'exec'), module.__dict__) + return module + + +def cpu_environment(): + require(sys.dont_write_bytecode and 'torch' not in sys.modules and 'runtime_light' not in sys.modules, + 'Use a fresh original Python -B process') + require(Path(sys.prefix) == ORIGINAL_ENV, 'Reference export requires the original registered .venv') + os.environ['CUDA_VISIBLE_DEVICES'] = '' + os.environ['PYTHONDONTWRITEBYTECODE'] = '1' + + +def run_cpu(argv, directory, label): + started = time.perf_counter() + environment = dict(os.environ, CUDA_VISIBLE_DEVICES='', PYTHONDONTWRITEBYTECODE='1') + with (directory / (label + '.stdout.log')).open('xb') as stdout, \ + (directory / (label + '.stderr.log')).open('xb') as stderr: + process = subprocess.run(argv, cwd=directory, env=environment, stdout=stdout, stderr=stderr, + timeout=1800, check=False) + elapsed = time.perf_counter() - started + save(directory / (label + '-process.json'), dict(argv=argv, exit_code=process.returncode, + wall_seconds=elapsed, CUDA_VISIBLE_DEVICES='', brain_executed=False)) + require(process.returncode == 0, label + ' failed; partial outputs retained') + return elapsed + + +def reference_worker(directory): + cpu_environment() + import numpy as np + directory = unlinked(directory) + inputs = load_npz(directory / 'inputs.npz') + command = read(directory / 'command.json') + sources = read(directory / 'reference-sources.json') + require(sources['conditioning_update_source']['sha256'] == U_SHA + and sources['eligibility_bridge_source']['sha256'] == BRIDGE_SHA + and sources['phase4j_source']['sha256'] == J_SHA, 'Original source pins differ') + for info in sources.values(): + check(info['path'], info['sha256']) + update = load_verified_source('_acceleration_original_U', sources['conditioning_update_source']['path'], U_SHA) + kwargs = {**inputs, **command, 'bridge_source': sources['eligibility_bridge_source']['path'], + 'phase4j_source': sources['phase4j_source']['path']} + before = {name: array_evidence(value) for name, value in inputs.items()} + started = time.perf_counter() + cold = update.update_epoch(**kwargs) + cold_seconds = time.perf_counter() - started + started = time.perf_counter() + warm = update.update_epoch(**kwargs) + warm_seconds = time.perf_counter() - started + require(canonical(cold.receipt) == canonical(warm.receipt), 'Cold/warm original receipts differ') + for name in OUTPUT_ARRAYS: + require(array_evidence(getattr(cold, name)) == array_evidence(getattr(warm, name)), + 'Cold/warm original output differs: ' + name) + require(before == {name: array_evidence(value) for name, value in inputs.items()}, 'Reference input mutated') + import torch + require(not torch.cuda.is_initialized() and os.environ['CUDA_VISIBLE_DEVICES'] == '', 'CUDA initialized in CPU reference') + arrays = {'reference_' + name: getattr(cold, name) for name in OUTPUT_ARRAYS} + commands = [] if command['us_tick'] is None else [command['us_tick']] + arrays['reference_command_steps'] = np.asarray(commands, dtype=np.int64) + arrays['reference_snapshots'] = (np.empty((0, len(inputs['kc'])), np.float32) if not commands + else cold.sampled_trace[None, :].copy()) + save_npz(directory / 'reference.npz', arrays) + save(directory / 'reference-update-receipt.json', cold.receipt) + save(directory / 'reference-timing.json', dict(schema='conditioning-original-cpu-timing-v1', + python_version=sys.version, executable=sys.executable, environment_prefix=sys.prefix, + numpy_version=np.__version__, torch_version=torch.__version__, cuda_initialized=False, + cold_update_seconds=cold_seconds, warm_update_seconds=warm_seconds, + timing_scope='original update_epoch API with loaded arrays, includes source checks/AST parsing; cold includes first Torch import; not pure kernel time', + process_time_scope='reference-process.json separately includes startup, loading, two updates and output serialization', + snapshot_scope='exact sole declared-US snapshot equals sampled_trace; empty snapshots for no US', + output_arrays={name: array_evidence(array) for name, array in arrays.items()}, + source_pins=sources, cold_warm_bitwise_exact=True, brain_executed=False, weights_installed=False)) + + +def validate_destination(output, protected): + output = unlinked(output) + require(not output.exists() and output.parent.is_dir(), 'Output must be a fresh directory with an existing parent') + require(not any(output.is_relative_to(unlinked(parent)) for parent in protected), + 'Output overlaps a frozen input/source directory') + return output + + +def export(output): + cpu_environment() + started = time.perf_counter() + own = identity(Path(__file__).absolute()) + regpath = EXP / 'protocol/preregistered.json' + check(regpath, REG_SHA) + reg = read(regpath) + require(reg['schema'] == 'normal-conditioning-registration-v1' + and reg['source_code_commit'] == SOURCE_COMMIT and reg['scientific_scope'] == 'normal-acquisition-only' + and reg['subsequent_assays_authorized'] is False, 'Wrong fixed AR authority') + source_pins = {**reg['source_sha256'], **{p: sha for p, sha in reg['input_sha256'].items() + if Path(p).is_relative_to(MODEL)}} + for path, sha in source_pins.items(): + check(path, sha) + planpath = EXP / 'protocol/normal-plan.json' + check(planpath, reg['prepared_sha256'][str(planpath)]) + plan = read(planpath) + entries, tracked, control_pins, snapshot = ledger_snapshot(EXP, reg) + selected, available = select_tasks(plan['tasks'], set(entries)) + tasks_by_id = {task['task_id']: task for task in plan['tasks']} + protected = {EXP, MODEL, *[Path(p).parent for p in reg['input_sha256']]} + output = validate_destination(output, protected) + output.mkdir() + selection = dict(schema='conditioning-acceleration-selection-v1', registration_sha256=REG_SHA, + source_code_commit=SOURCE_COMMIT, rule='first and last lexicographic task_id per arm among committed train brain/update/verify triples', + metrics_used_for_selection=False, available_train_counts=available, selected_tasks=selected, + selected_count=len(selected), maximum_cases=14, ledger_snapshot=snapshot, + coverage_limit='engineering fixtures only; neither complete scientific-condition coverage nor a learning result') + save(output / 'selection.json', selection) + save(output / 'original-source-pins.json', source_pins) + # No task result, update or event array is read before the immutable selection above. + fixtures = [] + for index, task in enumerate(selected): + item_started = time.perf_counter() + fixture_id = f'{index:02d}-{task["arm"]}' + directory = output / fixture_id + directory.mkdir() + key = task_key(task) + file_pins = {} + jobs = {} + process_records = {} + for stage in ('brain', 'update', 'verify'): + job, pins, process = validate_process(entries[key + '--' + stage], tracked, EXP) + jobs[stage] = job + file_pins.update(pins) + process_records[stage] = process + result_path = EXP / 'results' / (key + '.json') + file_pins[str(result_path)] = check(result_path, tracked[str(result_path)]) + result = read(result_path) + require(result['schema'] == 'normal-task-result-v1' and result['complete'] is True + and result['registration_sha256'] == REG_SHA and result['task'] == task, 'Selected result/plan differs') + authority_path = unlinked(result['epoch_authority']['path']) + authority_sha = result['epoch_authority']['sha256'] + file_pins[str(authority_path)] = check(authority_path, tracked[str(authority_path)]) + require(file_pins[str(authority_path)]['sha256'] == authority_sha, 'Authority result/ledger pin differs') + authority = read(authority_path) + command = dict(authority['command'], n_neurons=166700) + require(authority['epoch_id'] == task['task_id'] and authority['diagnostic_mode'] == 'light' + and command == dict(steps=task['steps'], n_neurons=166700, compartment=task['compartment_verbatim'], + us_tick=task['recipe']['us']['tick'], d_command=task['recipe']['D_command'], + frozen=not task['recipe']['update_enabled'], trace_backend='torch'), 'Actual command differs from registered task') + update_dir = unlinked(result['update_dir']) + require(update_dir == EXP / 'updates' / key and authority_path == EXP / 'jobs' / (key + '.authority.json') + and authority['checkpoint_dir'] == result['checkpoint_dir'] == str(EXP / 'runs' / key) + and authority['checkpoint_manifest_sha256'] == result['checkpoint_manifest_sha256'], + 'Task checkpoint/update location differs') + for stage in ('update', 'verify'): + require(jobs[stage]['authority_file'] == str(authority_path) and jobs[stage]['authority_sha256'] == authority_sha, + 'Authority not bound to selected successful job') + require(jobs['brain']['output_dir'] == authority['checkpoint_dir'] and jobs['update']['output_dir'] == str(update_dir) + and jobs['verify']['update_dir'] == str(update_dir), 'Selected process output differs') + require(jobs['brain']['kind'] == 'brain' and jobs['update']['kind'] == 'update' + and jobs['verify']['kind'] == 'cpu_verify', 'Selected process role differs') + dependency = task['delta_dependency'] + if dependency == dict(kind='zero', source=None): + expected_before = reg['initial_overlay'] + else: + require(dependency['kind'] == 'epoch_delta_after' and dependency['source'].endswith('/delta-after'), + 'Unknown registered overlay dependency') + previous = tasks_by_id[dependency['source'][:-len('/delta-after')]] + require(previous['kind'] == 'train' and all(previous[field] == task[field] for field in ('block_id', 'arm')) + and task_key(previous) + '--verify' in entries, 'Previous overlay is not a committed same-arm training output') + previous_dir = EXP / 'updates' / task_key(previous) + expected_before = dict(overlay_file=str(previous_dir / 'delta-after.npz'), + overlay_manifest=str(previous_dir / 'overlay-manifest.json')) + require(result['overlay_before'] == expected_before + and jobs['brain']['overlay_file'] == expected_before['overlay_file'] + and jobs['brain']['overlay_manifest'] == expected_before['overlay_manifest'] + and authority['files']['overlay_before']['path'] == expected_before['overlay_file'] + and authority['files']['overlay_manifest_before']['path'] == expected_before['overlay_manifest'], + 'Actual before overlay differs from registered branch dependency') + registered_files = {**reg['input_sha256'], **reg['prepared_sha256']} + for info in authority['files'].values(): + require(jobs['update']['input_sha256'].get(info['path']) == info['sha256'], 'Authority role absent from successful update inputs') + trusted_sha = registered_files.get(info['path'], tracked.get(info['path'], {}).get('sha256')) + require(trusted_sha == info['sha256'], 'Authority role lacks registered or committed file identity') + file_pins[info['path']] = check(info['path'], info['sha256']) + for name in ('chain-receipt.json', 'update-receipt.json', 'delta-after.npz', 'overlay-manifest.json'): + path = update_dir / name + file_pins[str(path)] = check(path, tracked[str(path)]) + chain_sha = file_pins[str(update_dir / 'chain-receipt.json')]['sha256'] + require(jobs['verify']['chain_sha256'] == chain_sha, 'Selected CPU validator used a different chain') + cpu_path = unlinked(result['cpu_validation']['path']) + file_pins[str(cpu_path)] = check(cpu_path, tracked[str(cpu_path)]) + require(file_pins[str(cpu_path)]['sha256'] == result['cpu_validation']['sha256'], 'Original CPU receipt pin differs') + original_validation = read(cpu_path) + require(original_validation['complete'] is True and original_validation['frozen_U_verify_receipt_recomputed'] is True + and original_validation['actual_sealed_events_used'] is True + and original_validation['full_candidate_mapping_verified'] is True + and original_validation['epoch_authority_sha256'] == authority_sha, 'Original CPU validation incomplete') + validator = MODEL / 'conditioning-execution-v1/validate_update_cpu.py' + check(validator, reg['input_sha256'][str(validator)]) + validator_seconds = run_cpu([sys.executable, '-B', str(validator), '--authority', str(authority_path), + '--authority-sha256', authority_sha, '--update-dir', str(update_dir), '--chain-sha256', chain_sha, + '--receipt', str(directory / 'fresh-cpu-validation.json')], directory, 'validation') + fresh = read(directory / 'fresh-cpu-validation.json') + require(fresh == original_validation, 'Fresh original validator receipt differs from committed receipt') + import numpy as np + paths = {role: unlinked(info['path']) for role, info in authority['files'].items()} + state = load_npz(paths['state']) + mapping = load_npz(paths['candidate_map']) + before = load_npz(paths['overlay_before']) + after = load_npz(update_dir / 'delta-after.npz') + arrays = dict(step=state['step'], neuron_idx=state['neuron_idx'], + kc=np.load(paths['kc'], allow_pickle=False), candidate_edge_id=np.load(paths['mask'], allow_pickle=False), + candidate_sources=mapping['source_neuron_idx'], candidate_targets=mapping['target_neuron_idx'], + candidate_compartments=mapping['target_compartment_verbatim'], delta_before=before['plastic_delta_mv'], + base_candidate=np.load(paths['base_candidate'], allow_pickle=False)) + require(len(arrays['kc']) == 4064 and len(arrays['candidate_edge_id']) == 14232, 'Complete identity dimensions differ') + save_npz(directory / 'inputs.npz', arrays) + save(directory / 'command.json', command) + sources = {role: authority['files'][role] for role in + ('conditioning_update_source', 'eligibility_bridge_source', 'phase4j_source')} + save(directory / 'reference-sources.json', sources) + reference_seconds = run_cpu([sys.executable, '-B', str(Path(__file__).absolute()), + '--reference-worker', str(directory)], directory, 'reference') + reference = load_npz(directory / 'reference.npz') + receipt = read(directory / 'reference-update-receipt.json') + stored_receipt = read(update_dir / 'update-receipt.json') + require(canonical(receipt) == canonical(stored_receipt), 'Original recomputation differs from stored update receipt') + require(array_evidence(reference['reference_delta_after']) == array_evidence(after['plastic_delta_mv']), + 'Original reference delta bytes differ from committed delta-after') + for name in OUTPUT_ARRAYS: + require(array_evidence(reference['reference_' + name]) == stored_receipt['outputs'][name], + 'Reference trace/delta does not match original receipt byte identity') + fixture = {**arrays, **reference, 'stored_delta_after': after['plastic_delta_mv']} + save_npz(directory / 'fixture.npz', fixture) + save(directory / 'authority.json', authority) + save(directory / 'stored-update-receipt.json', stored_receipt) + save(directory / 'source-file-pins.json', file_pins) + timing = read(directory / 'reference-timing.json') + record = dict(fixture_id=fixture_id, task=task, npz=f'{fixture_id}/fixture.npz', + command=f'{fixture_id}/command.json', npz_sha256=digest(directory / 'fixture.npz'), + command_sha256=digest(directory / 'command.json'), authority_sha256=authority_sha, + original_update_payload_sha256=receipt['payload_sha256'], + exact_stored_delta_and_trace_verified=True, full_canonical_KC_count=4064, full_candidate_edge_count=14232, + original_validator_verified_full_graph_mapping=True, reference_timing=timing, + reference_subprocess_wall_seconds=reference_seconds, original_validator_subprocess_wall_seconds=validator_seconds, + fixture_end_to_end_seconds=time.perf_counter() - item_started, + historical_process_seconds={stage: {field: proc[field] for field in ('child_wall_seconds', 'end_to_end_seconds')} + for stage, proc in process_records.items()}, + input_arrays={name: array_evidence(value) for name, value in arrays.items()}, + output_arrays={name: array_evidence(value) for name, value in reference.items()}) + save(directory / 'fixture.json', record) + fixtures.append(record) + print(canonical(dict(fixture=fixture_id, complete=True, task_id=task['task_id'])), flush=True) + for path, pin in control_pins.items(): + check(path, pin) + require(not (EXP / '.append-ledger/writer.lock').exists() + and len(list((EXP / '.append-ledger/entries').iterdir())) == snapshot['entries'], + 'Ledger changed during export; fixture set not published') + for path, sha in source_pins.items(): + check(path, sha) + check(Path(__file__).absolute(), own) + # Recheck all selected immutable input/output pins after the entire export. + all_pins = {} + for fixture in fixtures: + all_pins.update(read(output / fixture['fixture_id'] / 'source-file-pins.json')) + for path, pin in all_pins.items(): + check(path, pin) + manifest = dict(schema='conditioning-acceleration-fixtures-v1', complete=True, + registration_sha256=REG_SHA, source_code_commit=SOURCE_COMMIT, + exporter_source=dict(path=str(Path(__file__).absolute()), **own), + created_utc=datetime.now(timezone.utc).isoformat(), selection_sha256=digest(output / 'selection.json'), + fixtures=fixtures, fixture_count=len(fixtures), coverage=selection, + total_export_seconds=time.perf_counter() - started, + original_files_modified=False, brain_executed=False, training_executed=False, + weights_installed=False, learning_pass_claimed=False, + authority_limit='committed selected historical processes and original CPU reference only; AR remains incomplete', + files={str(path.relative_to(output)): identity(path) for path in sorted(output.rglob('*')) if path.is_file()}) + save(output / 'manifest.json', manifest) + return manifest + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument('--output', type=Path) + mode.add_argument('--reference-worker', type=Path, help=argparse.SUPPRESS) + args = parser.parse_args() + if args.reference_worker: + reference_worker(args.reference_worker) + else: + result = export(args.output) + print(canonical(dict(complete=True, output=str(args.output), fixtures=result['fixture_count'])), flush=True) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py b/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py new file mode 100644 index 0000000..3bfee1b --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py @@ -0,0 +1,242 @@ +"""Fixed synthetic full-mapping fixtures, using only the original CPU oracle. + +Run the original science Python -B with --template pointing to a completed real +export's fixture.npz and --output a new directory. Anatomical arrays retain the +real fixture's authenticated identity; events and delta-before are explicitly +synthetic. No brain, training, installed weights or original files are changed. +""" +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +from pathlib import Path +import sys +import time + +import export_fixtures as original + + +COMPARTMENTS = ("B2B'2a", "a1", "y2a'1", "y5B'2a") +STATIC_ARRAYS = ("kc", "candidate_edge_id", "candidate_sources", "candidate_targets", + "candidate_compartments", "base_candidate") +STEPS = 1200 +N_NEURONS = 166700 +SEEDS = (717, 20261001) + + +def template_binding(template, metadata=None, manifest_path=None): + """Bind to a completed real export and its original full-graph validation.""" + template = original.unlinked(template) + directory = template.parent + root = directory.parent + metadata = original.unlinked(metadata or directory / 'fixture.json') + manifest_path = original.unlinked(manifest_path or root / 'manifest.json') + original.require(template == directory / 'fixture.npz' and metadata == directory / 'fixture.json' + and manifest_path == root / 'manifest.json', 'Use one intact real-fixture export') + manifest = original.read(manifest_path) + original.require(manifest['schema'] == 'conditioning-acceleration-fixtures-v1' + and manifest['complete'] is True and manifest['registration_sha256'] == original.REG_SHA, + 'Template requires a completed real-fixture manifest') + required = ('fixture.npz', 'fixture.json', 'command.json', 'reference-sources.json', + 'fresh-cpu-validation.json', 'validation-process.json', 'reference-process.json') + pins = {str(manifest_path): original.identity(manifest_path)} + for name in required: + path = directory / name + rel = path.relative_to(root).as_posix() + original.require(rel in manifest['files'], 'Template file absent from manifest: ' + rel) + pins[str(path)] = original.check(path, manifest['files'][rel]) + record = original.read(metadata) + original.require(record in manifest['fixtures'] and record['fixture_id'] == directory.name + and record['npz'] == template.relative_to(root).as_posix() + and record['npz_sha256'] == pins[str(template)]['sha256'] + and record['exact_stored_delta_and_trace_verified'] is True + and record['original_validator_verified_full_graph_mapping'] is True + and record['full_canonical_KC_count'] == 4064 + and record['full_candidate_edge_count'] == 14232, + 'Template record does not bind an authenticated complete original mapping') + fresh = original.read(directory / 'fresh-cpu-validation.json') + original.require(fresh['complete'] is True and fresh['full_candidate_mapping_verified'] is True + and fresh['frozen_U_verify_receipt_recomputed'] is True + and fresh['epoch_authority_sha256'] == record['authority_sha256'], + 'Original full-graph CPU validation differs') + for name in ('validation-process.json', 'reference-process.json'): + process = original.read(directory / name) + original.require(type(process['exit_code']) is int and process['exit_code'] == 0 + and process['argv'][0] == str(original.ORIGINAL_ENV / 'bin/python') + and process['CUDA_VISIBLE_DEVICES'] == '', 'Template original CPU process invalid') + sources = original.read(directory / 'reference-sources.json') + expected = dict(conditioning_update_source=original.U_SHA, + eligibility_bridge_source=original.BRIDGE_SHA, phase4j_source=original.J_SHA) + original.require(set(sources) == set(expected), 'Original source roles differ') + for role, sha in expected.items(): + original.require(sources[role]['sha256'] == sha, 'Original source identity differs: ' + role) + pins[sources[role]['path']] = original.check(sources[role]['path'], sha) + arrays = original.load_npz(template) + for name in original.INPUT_ARRAYS: + original.require(original.array_evidence(arrays[name]) == record['input_arrays'][name], + 'Template input array identity differs: ' + name) + command = original.read(directory / 'command.json') + original.require(command['steps'] == STEPS and command['n_neurons'] == N_NEURONS + and command['trace_backend'] == 'torch', 'Unexpected original fixture dimensions') + return arrays, sources, pins, record + + +def fixed_cases(template): + """Create the declared sixteen cases without inspecting candidate outputs.""" + import numpy as np + kc = template['kc'] + sources = template['candidate_sources'] + base = template['base_candidate'] + lower = np.float32(-0.25) * base + original.require(kc.dtype == np.int32 and len(kc) == 4064 and np.all(kc[1:] > kc[:-1]) + and len(base) == len(template['candidate_edge_id']) == 14232, + 'Expected complete canonical anatomical arrays') + original.require(set(np.unique(template['candidate_compartments'])) == set(COMPARTMENTS), + 'Template must cover all four original compartments') + active_sources = np.unique(sources) + chosen = active_sources[:3] + original.require(len(chosen) == 3, 'Need three distinct candidate KC sources') + non_kc = int(np.flatnonzero(~np.isin(np.arange(N_NEURONS, dtype=np.int32), kc))[0]) + zero = np.zeros(len(base), np.float32) + partial = np.float32(-0.125) * base + partial[::3] = np.float32(0) + + def events(rows): + rows = sorted(set((int(t), int(n)) for t, n in rows)) + return (np.asarray([row[0] for row in rows], np.int32), + np.asarray([row[1] for row in rows], np.int32)) + + def case(name, rows, *, us=900, d=1., frozen=False, compartment=COMPARTMENTS[0], delta=None, tags=()): + tick, neuron = events(rows) + arrays = {key: template[key].copy() for key in STATIC_ARRAYS} + arrays.update(step=tick, neuron_idx=neuron, + delta_before=(zero if delta is None else delta).copy()) + return dict(name=name, arrays=arrays, + command=dict(steps=STEPS, n_neurons=N_NEURONS, compartment=compartment, + us_tick=us, d_command=d, frozen=frozen, trace_backend='torch'), + tags=list(tags), synthetic_events=True, synthetic_delta_before=True) + + boundary = [(899, chosen[0]), (900, chosen[1]), (901, chosen[2]), + (0, non_kc), (900, non_kc), (1199, non_kc)] + ends = [(0, chosen[0]), (1, chosen[1]), (1199, chosen[2]), (0, non_kc), (1199, non_kc)] + all_active = [(900, n) for n in active_sources] + cases = [ + case('empty-us900', [], tags=('empty', 'positive-zero')), + case('non-kc-only', [(0, non_kc), (900, non_kc), (1199, non_kc)], tags=('ignore-non-KC',)), + case('no-us-mixed', boundary, us=None, d=0., tags=('no-US', 'sampled-positive-zero', 'nonzero-final')), + case('us-first-tick', ends, us=0, tags=('US-zero', 'after-current-spike')), + case('us-final-tick', ends, us=1199, tags=('US-final', 'after-current-spike')), + case('before-on-after-us900', boundary, tags=('US-boundary', 'mixed-KC-non-KC')), + case('d-zero-existing-delta', boundary, d=0., delta=partial, tags=('D-zero', 'bitwise-identity')), + case('frozen-existing-delta', boundary, frozen=True, delta=partial, tags=('frozen', 'bitwise-identity')), + ] + for index, compartment in enumerate(COMPARTMENTS): + cases.append(case(f'compartment-{index}', all_active, compartment=compartment, delta=partial, + tags=('full-source-activation', 'off-compartment-bitwise-identity', compartment))) + cases.append(case('already-at-floor', all_active, delta=lower, tags=('floor', 'already-saturated'))) + near = np.nextafter(lower, np.full_like(lower, np.inf)) + cases.append(case('one-ulp-above-floor', all_active, delta=near, tags=('floor', 'new-saturation'))) + for seed in SEEDS: + rng = np.random.default_rng(seed) + ticks = rng.integers(0, STEPS, size=5000) + neurons = rng.choice(kc, size=5000) + extra_ticks = rng.integers(0, STEPS, size=1000) + extra_neurons = rng.integers(0, N_NEURONS, size=1000) + rows = list(zip(ticks, neurons)) + list(zip(extra_ticks, extra_neurons)) + delta = -(rng.uniform(0., .249, len(base)).astype(np.float32) * base) + delta[::7] = np.float32(0) + delta[1::11] = lower[1::11] + item = case(f'seeded-sparse-{seed}', rows, delta=delta, tags=('seeded-sparse', 'mixed-delta', 'canonical-zero')) + item['rng_seed'] = seed + cases.append(item) + original.require(len(cases) == 16, 'Fixed synthetic coverage changed') + for item in cases: + for name in STATIC_ARRAYS: + original.require(original.array_evidence(item['arrays'][name]) == original.array_evidence(template[name]), + 'Synthetic generation changed an anatomical array: ' + name) + original.require(not np.any((item['arrays']['delta_before'] == 0) + & np.signbit(item['arrays']['delta_before'])), 'Synthetic delta has negative zero') + return cases + + +def export(template, output, metadata=None, manifest_path=None): + original.cpu_environment() + started = time.perf_counter() + own = original.identity(Path(__file__).absolute()) + helper = original.identity(Path(original.__file__).absolute()) + arrays, sources, pins, template_record = template_binding(template, metadata, manifest_path) + cases = fixed_cases(arrays) + protected = {original.ROOT, Path(template).parent.parent, Path(__file__).absolute().parent, + *(Path(info['path']).parent for info in sources.values())} + output = original.validate_destination(output, protected) + output.mkdir() + original.save(output / 'template-binding.json', dict(template=str(template), template_record=template_record, + input_pins=pins, scope='complete anatomical identity inherited from original full-graph CPU validation', + original_events_reused=False, original_delta_before_reused=False)) + original.save(output / 'selection.json', dict(schema='conditioning-fixed-synthetic-selection-v1', + names=[item['name'] for item in cases], count=16, seeds=list(SEEDS), + chosen_before_candidate_execution=True, original_full_mapping_preserved=True)) + fixtures = [] + for index, item in enumerate(cases): + fixture_id = f'{index:02d}-{item["name"]}' + directory = output / fixture_id + directory.mkdir() + original.save_npz(directory / 'inputs.npz', item['arrays']) + original.save(directory / 'command.json', item['command']) + original.save(directory / 'reference-sources.json', sources) + seconds = original.run_cpu([sys.executable, '-B', str(Path(original.__file__).absolute()), + '--reference-worker', str(directory)], directory, 'reference') + reference = original.load_npz(directory / 'reference.npz') + receipt = original.read(directory / 'reference-update-receipt.json') + timing = original.read(directory / 'reference-timing.json') + original.require(timing['numpy_version'] == template_record['reference_timing']['numpy_version'] + and timing['torch_version'] == template_record['reference_timing']['torch_version'], + 'Synthetic oracle differs from the real fixture original environment') + for name in original.OUTPUT_ARRAYS: + original.require(original.array_evidence(reference['reference_' + name]) == receipt['outputs'][name], + 'Synthetic reference output/receipt differs: ' + name) + original.save_npz(directory / 'fixture.npz', {**item['arrays'], **reference}) + record = dict(fixture_id=fixture_id, kind='synthetic', name=item['name'], tags=item['tags'], + npz=f'{fixture_id}/fixture.npz', command=f'{fixture_id}/command.json', + npz_sha256=original.digest(directory / 'fixture.npz'), + command_sha256=original.digest(directory / 'command.json'), + original_update_payload_sha256=receipt['payload_sha256'], + exact_stored_delta_and_trace_verified=False, historical_scientific_output_exists=False, + full_canonical_KC_count=4064, full_candidate_edge_count=14232, + original_validator_verified_full_graph_mapping=True, + mapping_identity_scope='inherited unchanged from template; events and delta-before are synthetic', + reference_timing=timing, reference_subprocess_wall_seconds=seconds, + input_arrays={name: original.array_evidence(value) for name, value in item['arrays'].items()}, + output_arrays={name: original.array_evidence(value) for name, value in reference.items()}) + original.save(directory / 'fixture.json', record) + fixtures.append(record) + print(original.canonical(dict(fixture=fixture_id, kind='synthetic', complete=True)), flush=True) + for path, pin in pins.items(): + original.check(path, pin) + original.check(Path(__file__).absolute(), own) + original.check(Path(original.__file__).absolute(), helper) + manifest = dict(schema='conditioning-acceleration-fixtures-v1', kind='synthetic', complete=True, + registration_sha256=original.REG_SHA, source_code_commit=original.SOURCE_COMMIT, + exporter_source=dict(path=str(Path(__file__).absolute()), **own), + original_helper_source=dict(path=str(Path(original.__file__).absolute()), **helper), + created_utc=datetime.now(timezone.utc).isoformat(), fixtures=fixtures, fixture_count=len(fixtures), + selection_sha256=original.digest(output / 'selection.json'), + template_binding_sha256=original.digest(output / 'template-binding.json'), + total_export_seconds=time.perf_counter() - started, original_files_modified=False, + brain_executed=False, training_executed=False, weights_installed=False, learning_pass_claimed=False, + authority_limit='synthetic arithmetic edge cases with original mapping; not historical scientific trials', + files={str(path.relative_to(output)): original.identity(path) + for path in sorted(output.rglob('*')) if path.is_file()}) + original.save(output / 'manifest.json', manifest) + return manifest + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--template', type=Path, required=True) + parser.add_argument('--metadata', type=Path) + parser.add_argument('--manifest', type=Path) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + result = export(args.template, args.output, args.metadata, args.manifest) + print(original.canonical(dict(complete=True, output=str(args.output), fixtures=result['fixture_count'])), flush=True) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/gpu_backend.py b/models/male-cns-lif/conditioning-acceleration-v1/gpu_backend.py new file mode 100644 index 0000000..dec94db --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/gpu_backend.py @@ -0,0 +1,458 @@ +"""Independent CUDA eligibility candidate; never installs weights or runs a brain. + +The caller authenticates frozen sources, configuration, the complete KC list, +and the candidate edge mapping. This module does not infer D from DAN activity. +CuPy is imported only when CudaEligibility is instantiated. Set CUPY_CACHE_DIR +to a dedicated candidate directory before doing so; this module writes no files. + +API: prepare_events(...) once, then CudaEligibility(...).run(prepared, ...). +All commands are sampled AFTER the current tick; only the LAST command snapshot +feeds the single epoch-boundary update. With no commands sampled_trace is +0. +cpu_reference has the same run arguments for host-only validation, and is not a +replacement for comparison against the SHA-pinned frozen implementation. +""" +from __future__ import annotations + +from dataclasses import dataclass +import math +import os +from pathlib import Path +import time +from typing import Any + +import numpy as np + + +COMPARTMENTS = ("B2B'2a", "a1", "y2a'1", "y5B'2a") +_INT32_MAX = np.iinfo(np.int32).max +CUDA_OPTIONS = ("--std=c++11", "--fmad=false", "--ftz=false") +CUDA_SOURCE = r''' +extern "C" __global__ +void eligibility_trace(const long long* ptr, const int* ticks, + const int* commands, const int n_commands, + const int n_kc, const int n_steps, const float decay, + float* snapshots, float* final_trace, float* sampled) { + const int k = blockDim.x * blockIdx.x + threadIdx.x; + if (k >= n_kc) return; + long long cursor = ptr[k]; + const long long end = ptr[k + 1]; + int command = 0; + float trace = 0.0f; + float last_sample = 0.0f; + // Each KC follows the FULL sequential recurrence, including empty ticks. + for (int tick = 0; tick < n_steps; ++tick) { + float spike = 0.0f; + if (cursor < end && ticks[cursor] == tick) { + spike = 1.0f; + ++cursor; + } + trace = __fmul_rn(trace, decay); + trace = __fadd_rn(trace, spike); + trace = trace < 1.0f ? trace : 1.0f; + if (command < n_commands && commands[command] == tick) { + snapshots[(long long)command * n_kc + k] = trace; + last_sample = trace; + ++command; + } + } + final_trace[k] = trace; + sampled[k] = last_sample; +} + +extern "C" __global__ +void eligibility_delta(const float* sampled, const int* source_local, + const unsigned char* selected, const float* delta, + const float* base, const int n_edges, + const float negative_eta_times_d, + const float negative_floor, const int nonzero_d, + const int frozen, float* output) { + const int edge = blockDim.x * blockIdx.x + threadIdx.x; + if (edge >= n_edges) return; + if (frozen) { + output[edge] = delta[edge]; + return; + } + float update = 0.0f; + if (nonzero_d && selected[edge]) + update = __fmul_rn(negative_eta_times_d, sampled[source_local[edge]]); + const float out = __fadd_rn(delta[edge], update); + const float lower = __fmul_rn(negative_floor, base[edge]); + // np.maximum chooses its SECOND operand on equal finite values, including + // +0/-0. CUDA fmaxf is deliberately not substituted for that tie behavior. + output[edge] = out > lower ? out : lower; +} +''' + + +def _require(condition: Any, message: str) -> None: + if not condition: + raise ValueError(message) + + +def _integer(value: Any, name: str, minimum: int = 1) -> int: + _require(isinstance(value, (int, np.integer)) and not isinstance(value, (bool, np.bool_)), + f"{name} must be an integer") + _require(minimum <= int(value) <= _INT32_MAX, f"{name} outside supported int32 range") + return int(value) + + +def _array(value: Any, dtype: Any, name: str) -> np.ndarray: + _require(isinstance(value, np.ndarray) and value.ndim == 1 and value.dtype == np.dtype(dtype), + f"{name} must be a one-dimensional {np.dtype(dtype)} ndarray") + return value + + +def _owned(value: np.ndarray) -> np.ndarray: + result = np.array(value, copy=True, order="C") + result.setflags(write=False) + return result + + +def _scalar(value: Any, name: str) -> float: + _require(isinstance(value, (float, int, np.floating, np.integer)) + and not isinstance(value, (bool, np.bool_)), f"{name} must be a real scalar") + try: + result = float(value) + except (ValueError, OverflowError) as exc: + raise ValueError(f"{name} must be a finite scalar") from exc + _require(math.isfinite(result), f"{name} must be finite") + return result + + +@dataclass(frozen=True) +class PreparedEvents: + kc: np.ndarray + indptr: np.ndarray + ticks: np.ndarray + command_steps: np.ndarray + steps: int + n_neurons: int + input_event_count: int + preparation_seconds: float + + @property + def kc_event_count(self) -> int: + return len(self.ticks) + + +@dataclass(frozen=True) +class EligibilityResult: + sampled_trace: np.ndarray + final_trace: np.ndarray + snapshots: np.ndarray + delta_after: np.ndarray + timing: dict[str, Any] + audit: dict[str, Any] + + +def prepare_events(step: np.ndarray, neuron_idx: np.ndarray, kc: np.ndarray, + steps: int, command_steps: Any, *, n_neurons: int = 166700) -> PreparedEvents: + """Validate original sparse events and pack sorted per-KC event CSR on CPU. + + Inputs must be sorted by (tick, global neuron ID), unique and int32. No + [steps, neurons] raster is allocated. Original input storage is not retained. + Empty, one, or multiple sorted unique command ticks are supported. + """ + started = time.perf_counter() + steps = _integer(steps, "steps") + n_neurons = _integer(n_neurons, "n_neurons") + step = _array(step, np.int32, "step") + neuron_idx = _array(neuron_idx, np.int32, "neuron_idx") + kc = _array(kc, np.int32, "kc") + _require(0 < len(kc) <= _INT32_MAX and np.all((kc >= 0) & (kc < n_neurons)), + "KC IDs empty or outside graph") + _require(np.all(kc[1:] > kc[:-1]), "KC IDs must be sorted unique global IDs") + _require(len(step) == len(neuron_idx), "event columns have different lengths") + _require(np.all((step >= 0) & (step < steps)), "event tick outside trial") + _require(np.all((neuron_idx >= 0) & (neuron_idx < n_neurons)), "event neuron outside graph") + _require(np.all((step[1:] > step[:-1]) | + ((step[1:] == step[:-1]) & (neuron_idx[1:] > neuron_idx[:-1]))), + "events must be sorted by tick/global ID and unique") + commands = np.asarray(command_steps) + _require(commands.ndim == 1, "command_steps must be one-dimensional") + if commands.size == 0: + commands = np.empty(0, np.int32) + _require(commands.dtype.kind in "iu", "command_steps must be integers") + _require(np.all((commands >= 0) & (commands < steps)), "command tick outside trial") + _require(np.all(commands[1:] > commands[:-1]), "command_steps must be sorted unique") + local = np.searchsorted(kc, neuron_idx) + keep = (local < len(kc)) & (kc[np.minimum(local, len(kc) - 1)] == neuron_idx) + local = local[keep] + order = np.argsort(local, kind="stable") + ticks = step[keep][order] + counts = np.bincount(local, minlength=len(kc)).astype(np.int64, copy=False) + ptr = np.empty(len(kc) + 1, np.int64) + ptr[0] = 0 + np.cumsum(counts, out=ptr[1:]) + return PreparedEvents(_owned(kc), _owned(ptr), _owned(ticks), + _owned(commands.astype(np.int32)), steps, n_neurons, + len(step), time.perf_counter() - started) + + +def _validate_prepared(prepared: PreparedEvents) -> None: + _require(isinstance(prepared, PreparedEvents), "prepared must come from prepare_events") + _integer(prepared.steps, "steps") + _integer(prepared.n_neurons, "n_neurons") + kc = _array(prepared.kc, np.int32, "prepared.kc") + ptr = _array(prepared.indptr, np.int64, "prepared.indptr") + ticks = _array(prepared.ticks, np.int32, "prepared.ticks") + commands = _array(prepared.command_steps, np.int32, "prepared.command_steps") + _require(0 < len(kc) <= _INT32_MAX and np.all((kc >= 0) & (kc < prepared.n_neurons)) + and np.all(kc[1:] > kc[:-1]), "invalid prepared KC IDs") + _require(len(ptr) == len(kc) + 1 and ptr[0] == 0 and ptr[-1] == len(ticks) + and np.all(ptr[1:] >= ptr[:-1]), "invalid prepared CSR offsets") + _require(np.all((ticks >= 0) & (ticks < prepared.steps)), "invalid prepared event ticks") + # A decrease/equality is permitted only at a CSR row boundary. Checking all + # adjacent pairs stays vectorized and also catches a forged PreparedEvents. + nonincreasing = ticks[1:] <= ticks[:-1] + if len(ticks) > 1: + boundaries = ptr[1:-1] + boundaries = boundaries[(boundaries > 0) & (boundaries < len(ticks))] - 1 + nonincreasing[boundaries] = False + _require(not np.any(nonincreasing), "prepared KC ticks must be sorted unique") + _require(np.all((commands >= 0) & (commands < prepared.steps)) + and np.all(commands[1:] > commands[:-1]), "invalid prepared command ticks") + + +def _validate_run(prepared, candidate_sources, candidate_compartments, compartment, + delta_before, base_candidate, d, frozen, eta_mv, floor_fraction, decay): + _validate_prepared(prepared) + sources = _array(candidate_sources, np.int32, "candidate_sources") + delta = _array(delta_before, np.float32, "delta_before") + base = _array(base_candidate, np.float32, "base_candidate") + comps = candidate_compartments + _require(isinstance(comps, np.ndarray) and comps.ndim == 1 and comps.dtype.kind == "U", + "candidate_compartments must be a Unicode vector") + _require(len(delta) <= _INT32_MAX and len(sources) == len(delta) == len(base) == len(comps), + "candidate columns have different lengths or exceed int32 capacity") + _require(compartment in COMPARTMENTS and np.all(np.isin(comps, COMPARTMENTS)), + "unrecognized exact frozen compartment") + local = np.searchsorted(prepared.kc, sources) + _require(np.all(local < len(prepared.kc)), "all candidate sources must be canonical KCs") + _require(np.all(prepared.kc[local] == sources), "all candidate sources must be canonical KCs") + d = _scalar(d, "D") + _require(d in (0.0, 1.0), "D must be the fixed artificial scalar 0 or 1") + _require(isinstance(frozen, (bool, np.bool_)), "frozen must be Boolean") + eta = _scalar(eta_mv, "eta_mv") + floor = _scalar(floor_fraction, "floor_fraction") + decay_value = _scalar(decay, "decay") + _require(0 < eta <= np.finfo(np.float32).max, "eta_mv must be positive finite float32 representable") + _require(0 < floor <= 1, "floor_fraction must lie in (0,1]") + _require(0 < decay_value <= 1, "decay must lie in (0,1]") + _require(np.float32(eta) > 0 and np.float32(floor) > 0 and np.float32(decay_value) > 0, + "rule scalar underflows float32") + _require(np.all(np.isfinite(base)) and np.all(base > 0), + "original base must be finite positive float32 mV") + lower = np.float32(-floor) * base + _require(np.all(np.isfinite(delta)) and np.all((delta >= lower) & (delta <= 0)), + "delta violates original frozen floor") + _require(not np.any((delta == 0) & np.signbit(delta)), "delta zero must be canonical +0") + return dict(local=np.array(local, dtype=np.int32, order="C"), + selected=np.asarray(comps == compartment, dtype=np.uint8), + delta=np.array(delta, copy=True, order="C"), + base=np.array(base, copy=True, order="C"), d=d, frozen=bool(frozen), + eta=eta, floor=floor, decay=decay_value, + negative_eta_times_d=np.float32(-eta * d), negative_floor=np.float32(-floor)) + + +def _audit(prepared: PreparedEvents, work: dict, backend: str) -> dict[str, Any]: + return dict(backend=backend, steps=prepared.steps, kc_count=len(prepared.kc), + events=prepared.input_event_count, actual_kc_events=prepared.kc_event_count, + command_count=len(prepared.command_steps), candidate_count=len(work["base"]), + d_command=work["d"], frozen=work["frozen"], eta_mv=work["eta"], + floor_fraction=work["floor"], decay_python_float=work["decay"], + decay_float32_bits=int(np.asarray(np.float32(work["decay"])).view(np.uint32)), + initial_trace="float32 canonical +0", sampling="after_current_tick_spike", + update_sample="last command snapshot, or canonical +0 without commands", + trace_tick_skipping=False, dense_tick_neuron_raster=False, + original_source_executed=False, caller_must_verify_source_identity=True, + source_identity_verified_by_this_module=False, + complete_mapping_verified_by_this_module=False, + d_semantics="artificial scheduled command; not measured DAN activity", + actual_dan_activity_used_to_gate=False, brain_dynamics_executed=False, + training_executed=False, weights_installed=False, learning_pass_claimed=False) + + +def cpu_reference(prepared: PreparedEvents, candidate_sources: np.ndarray, + candidate_compartments: np.ndarray, compartment: str, + delta_before: np.ndarray, base_candidate: np.ndarray, *, d: float, + frozen: bool, eta_mv: float, floor_fraction: float, decay: float) -> EligibilityResult: + """Host arithmetic mirror for tests; frozen-source validation is external.""" + started = time.perf_counter() + work = _validate_run(prepared, candidate_sources, candidate_compartments, compartment, + delta_before, base_candidate, d, frozen, eta_mv, floor_fraction, decay) + nkc = len(prepared.kc) + trace = np.zeros(nkc, np.float32) + snapshots = np.empty((len(prepared.command_steps), nkc), np.float32) + positions = prepared.indptr[:-1].copy() + ends = prepared.indptr[1:] + spike = np.zeros(nkc, np.float32) + command = 0 + for tick in range(prepared.steps): + spike.fill(0) + active = np.flatnonzero(positions < ends) + active = active[prepared.ticks[positions[active]] == tick] + spike[active] = np.float32(1) + positions[active] += 1 + np.multiply(trace, np.float32(work["decay"]), out=trace) + np.add(trace, spike, out=trace) + np.minimum(trace, np.float32(1), out=trace) + if command < len(prepared.command_steps) and prepared.command_steps[command] == tick: + snapshots[command] = trace + command += 1 + sampled = snapshots[-1].copy() if len(snapshots) else np.zeros(nkc, np.float32) + output = work["delta"].copy() + if not work["frozen"]: + update = np.zeros(len(output), np.float32) + selected = work["selected"].astype(bool) + if work["d"] != 0: + update[selected] = work["negative_eta_times_d"] * sampled[work["local"][selected]] + output = output + update + output = np.maximum(output, work["negative_floor"] * work["base"]).astype(np.float32) + return EligibilityResult(_owned(sampled), _owned(trace), _owned(snapshots), _owned(output), + dict(end_to_end_seconds=time.perf_counter() - started, + preparation_seconds=prepared.preparation_seconds), + _audit(prepared, work, "numpy-host-math-mirror")) + + +class CudaEligibility: + """CuPy RawKernel candidate with explicit round-to-nearest float32 math. + + Use a dedicated environment and set an absolute CUPY_CACHE_DIR outside any + frozen tree before construction. The first run explicitly compiles/loads both + kernels. cold_compile_seconds measures that first-use compile/cache-load, + and does not prove an empty disk cache. Construction time is reported apart. + run includes validation, transfers and output ownership in end-to-end wall. + """ + + def __init__(self, device_index: int = 0): + started = time.perf_counter() + self.device_index = _integer(device_index, "device_index", 0) + cache_dir = os.environ.get("CUPY_CACHE_DIR", "") + _require(bool(cache_dir) and Path(cache_dir).is_absolute(), + "set CUPY_CACHE_DIR to an absolute candidate cache directory outside frozen trees") + try: + import cupy as cp + except ImportError as exc: + raise RuntimeError("CuPy unavailable; use the dedicated CUDA candidate environment") from exc + self.cp = cp + self.device = cp.cuda.Device(self.device_index) + with self.device: + self.stream = cp.cuda.Stream(non_blocking=True) + self.trace_kernel = cp.RawKernel(CUDA_SOURCE, "eligibility_trace", options=CUDA_OPTIONS) + self.delta_kernel = cp.RawKernel(CUDA_SOURCE, "eligibility_delta", options=CUDA_OPTIONS) + self.device.synchronize() + self.initialization_seconds = time.perf_counter() - started + self._compiled = False + + def run(self, prepared: PreparedEvents, candidate_sources: np.ndarray, + candidate_compartments: np.ndarray, compartment: str, + delta_before: np.ndarray, base_candidate: np.ndarray, *, d: float, + frozen: bool, eta_mv: float, floor_fraction: float, decay: float, + kernel_warmup: int = 0, kernel_repeats: int = 1) -> EligibilityResult: + """Transfer one fixture and optionally benchmark resident kernel repeats. + + Each repeat starts from zero trace and the SAME delta_before; repeats + never accumulate training updates. CUDA timing fields are medians across + kernel_repeats. end_to_end_seconds includes all warmups and repeats, so + use default counts for a single-call end-to-end benchmark. + """ + started = time.perf_counter() + work = _validate_run(prepared, candidate_sources, candidate_compartments, compartment, + delta_before, base_candidate, d, frozen, eta_mv, floor_fraction, decay) + kernel_warmup = _integer(kernel_warmup, "kernel_warmup", 0) + kernel_repeats = _integer(kernel_repeats, "kernel_repeats") + validation_seconds = time.perf_counter() - started + cp = self.cp + first_run = not self._compiled + compile_seconds = 0.0 + with self.device, self.stream: + if first_run: + compile_started = time.perf_counter() + self.trace_kernel.compile() + self.delta_kernel.compile() + self.stream.synchronize() + compile_seconds = time.perf_counter() - compile_started + self._compiled = True + transfer_started = time.perf_counter() + ptr_gpu, ticks_gpu, commands_gpu, local_gpu, selected_gpu, delta_gpu, base_gpu = [ + cp.asarray(a) for a in (prepared.indptr, prepared.ticks, prepared.command_steps, + work["local"], work["selected"], work["delta"], work["base"])] + nkc, nedges = len(prepared.kc), len(work["delta"]) + snapshots_gpu = cp.empty((len(prepared.command_steps), nkc), cp.float32) + final_gpu = cp.empty(nkc, cp.float32) + sampled_gpu = cp.empty(nkc, cp.float32) + output_gpu = cp.empty(nedges, cp.float32) + self.stream.synchronize() + upload_seconds = time.perf_counter() - transfer_started + trace_args = ( + ptr_gpu, ticks_gpu, commands_gpu, np.int32(len(prepared.command_steps)), + np.int32(nkc), np.int32(prepared.steps), np.float32(work["decay"]), + snapshots_gpu, final_gpu, sampled_gpu) + delta_args = ( + sampled_gpu, local_gpu, selected_gpu, delta_gpu, base_gpu, + np.int32(nedges), work["negative_eta_times_d"], work["negative_floor"], + np.int32(work["d"] != 0), np.int32(work["frozen"]), output_gpu) + + def launch_pair(): + self.trace_kernel(((nkc + 127) // 128,), (128,), trace_args) + if nedges: + self.delta_kernel(((nedges + 255) // 256,), (256,), delta_args) + + for _ in range(kernel_warmup): + launch_pair() + self.stream.synchronize() + resident_samples = [] + for _ in range(kernel_repeats): + trace_start, trace_end, update_end = cp.cuda.Event(), cp.cuda.Event(), cp.cuda.Event() + trace_start.record(self.stream) + self.trace_kernel(((nkc + 127) // 128,), (128,), trace_args) + trace_end.record(self.stream) + if nedges: + self.delta_kernel(((nedges + 255) // 256,), (256,), delta_args) + update_end.record(self.stream) + update_end.synchronize() + resident_samples.append(dict( + trace_kernel_seconds=cp.cuda.get_elapsed_time(trace_start, trace_end) / 1000.0, + update_kernel_seconds=cp.cuda.get_elapsed_time(trace_end, update_end) / 1000.0, + kernel_seconds=cp.cuda.get_elapsed_time(trace_start, update_end) / 1000.0)) + medians = {key: float(np.median([sample[key] for sample in resident_samples])) + for key in resident_samples[0]} + download_started = time.perf_counter() + sampled, final, snapshots, output = [ + _owned(cp.asnumpy(a, stream=self.stream, blocking=True)) + for a in (sampled_gpu, final_gpu, snapshots_gpu, output_gpu)] + download_seconds = time.perf_counter() - download_started + guard_started = time.perf_counter() + for name, array in (("sampled_trace", sampled), ("final_trace", final), ("snapshots", snapshots)): + if not (np.all(np.isfinite(array)) and np.all((array >= 0) & (array <= 1))): + raise RuntimeError(f"CUDA output guard failed: invalid {name}") + lower = work["negative_floor"] * work["base"] + if not (np.all(np.isfinite(output)) and np.all((output >= lower) & (output <= 0))): + raise RuntimeError("CUDA output guard failed: delta outside finite frozen floor") + selected = work["selected"].astype(bool) + if output[~selected].tobytes() != work["delta"][~selected].tobytes(): + raise RuntimeError("CUDA output guard failed: off-compartment delta changed") + if (work["frozen"] or work["d"] == 0) and output.tobytes() != work["delta"].tobytes(): + raise RuntimeError("CUDA output guard failed: frozen/zero-D identity") + guard_seconds = time.perf_counter() - guard_started + timing = dict(first_run=first_run, initialization_seconds=self.initialization_seconds, + cold_compile_seconds=compile_seconds, + compile_timing_semantics="first-use compile/cache-load; cache coldness externally controlled", + preparation_seconds=prepared.preparation_seconds, + run_validation_seconds=validation_seconds, + host_to_device_and_allocation_seconds=upload_seconds, + **medians, device_to_host_seconds=download_seconds, + output_guard_seconds=guard_seconds, + resident_kernel_warmup=kernel_warmup, + resident_kernel_repeats=kernel_repeats, + resident_kernel_samples=resident_samples, + kernel_statistic="median of resident timed repeats", + end_to_end_semantics="all requested warmups/repeats and transfers; preparation excluded", + end_to_end_seconds=time.perf_counter() - started) + audit = _audit(prepared, work, "cupy-rawkernel-independent-candidate") + audit.update(numpy_version=str(np.__version__), cupy_version=str(cp.__version__), + device_index=self.device_index, cuda_options=list(CUDA_OPTIONS), + float32_separate_rounding_intrinsics=True, tf32_used=False, + numerical_equivalence_established_by_this_call=False) + return EligibilityResult(sampled, final, snapshots, output, timing, audit) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/repair_environment.py b/models/male-cns-lif/conditioning-acceleration-v1/repair_environment.py new file mode 100644 index 0000000..fdc3f29 --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/repair_environment.py @@ -0,0 +1,22 @@ +"""Log repair of candidate-only packaging tool; preserve first failed attempt.""" +import json +import os +from pathlib import Path +import subprocess + +OUT = Path('/home/qishizhe/phase4ar-v100-acceleration-20261001') +PYTHON = OUT / '.venv/bin/python' +assert PYTHON.exists() and (OUT / 'install-candidate.stderr.log').exists() + +def run(args, stem): + with (OUT / (stem + '.stdout.log')).open('x') as stdout, (OUT / (stem + '.stderr.log')).open('x') as stderr: + result = subprocess.run([str(PYTHON), '-B', '-m', 'pip', *args], stdout=stdout, stderr=stderr, + env={**os.environ, 'PYTHONDONTWRITEBYTECODE': '1'}) + print(json.dumps({'step': stem, 'exit_code': result.returncode}), flush=True) + if result.returncode: + raise RuntimeError(stem) + +run(['install', '--disable-pip-version-check', 'pip==25.2'], 'upgrade-candidate-pip') +run(['install', '--disable-pip-version-check', '--report', str(OUT / 'pip-install-report.json'), + 'numpy==2.2.6', 'cupy-cuda12x==13.6.0'], 'install-candidate-attempt-02') +run(['freeze'], 'candidate-freeze') diff --git a/models/male-cns-lif/conditioning-acceleration-v1/requirements-candidate.txt b/models/male-cns-lif/conditioning-acceleration-v1/requirements-candidate.txt new file mode 100644 index 0000000..7d891ba --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/requirements-candidate.txt @@ -0,0 +1,2 @@ +numpy==2.2.6 +cupy-cuda12x==13.6.0 diff --git a/models/male-cns-lif/conditioning-acceleration-v1/setup_environment.py b/models/male-cns-lif/conditioning-acceleration-v1/setup_environment.py new file mode 100644 index 0000000..5aa7c1b --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/setup_environment.py @@ -0,0 +1,33 @@ +"""Provision only a new worktree and independent benchmark environment.""" +import json +import os +from pathlib import Path +import subprocess +import sys +from datetime import datetime, timezone + +ROOT=Path('/data/digital-drosophila') +WORK=Path('/home/qishizhe/digital-drosophila-v100-acceleration') +OUT=Path('/home/qishizhe/phase4ar-v100-acceleration-20261001') +BASE='09187e4d4d67badf1b418c5f211258ca5d8fb328' + +def run(argv, log): + with (OUT/(log+'.stdout.log')).open('x') as out, (OUT/(log+'.stderr.log')).open('x') as err: + r=subprocess.run(argv,stdout=out,stderr=err,env={**os.environ,'PYTHONDONTWRITEBYTECODE':'1'}) + print(json.dumps({'step':log,'exit_code':r.returncode}),flush=True) + if r.returncode:raise RuntimeError('Environment preparation failed: '+log) + +assert not WORK.exists() and not OUT.exists(), 'Provisioning targets must be new' +assert subprocess.check_output(['git','-C',str(ROOT),'rev-parse','HEAD'],text=True).strip()==BASE +assert subprocess.check_output(['git','-C',str(ROOT),'status','--porcelain'],text=True)=='' +OUT.mkdir() +with (OUT/'environment-authority.json').open('x') as f: + json.dump(dict(scope='independent engineering replay only',base_commit=BASE, + created_utc=datetime.now(timezone.utc).isoformat(),original_science_environment_modified=False, + original_experiment_resumed=False),f,indent=2) +run(['git','-C',str(ROOT),'worktree','add','-b','codex/v100-conditioning-acceleration',str(WORK),BASE],'create-worktree') +run([sys.executable,'-m','venv',str(OUT/'.venv')],'create-venv') +run([str(OUT/'.venv/bin/python'),'-B','-m','pip','install','--disable-pip-version-check', + '--report',str(OUT/'pip-install-report.json'),'numpy==2.2.6','cupy-cuda12x==13.6.0'],'install-candidate') +run([str(OUT/'.venv/bin/python'),'-B','-m','pip','freeze'],'candidate-freeze') +print(json.dumps({'worktree':str(WORK),'benchmark_root':str(OUT),'environment_ready':True}),flush=True) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/test_export_fixtures.py b/models/male-cns-lif/conditioning-acceleration-v1/test_export_fixtures.py new file mode 100644 index 0000000..3a439ba --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/test_export_fixtures.py @@ -0,0 +1,135 @@ +"""Small authority/selection tests; never access or run the production brain.""" +import importlib.util +from pathlib import Path +import tempfile +import unittest + + +PATH = Path(__file__).with_name('export_fixtures.py') +SPEC = importlib.util.spec_from_file_location('acceleration_fixture_export_tests', PATH) +E = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(E) + + +def task(arm, number, kind='train'): + return dict(task_id=f'block/normal/{arm}/{number:03d}', arm=arm, kind=kind) + + +def committed(tasks): + return {E.task_key(t) + '--' + stage for t in tasks for stage in ('brain', 'update', 'verify')} + + +class SelectionTests(unittest.TestCase): + def test_fixed_first_last_all_seven_arms_before_metrics(self): + tasks = [task(arm, n) for arm in E.ARMS for n in range(5)] + selected, counts = E.select_tasks(list(reversed(tasks)), committed(tasks)) + self.assertEqual(len(selected), 14) + self.assertEqual(counts, {arm: 5 for arm in E.ARMS}) + self.assertEqual([(t['arm'], t['task_id'].split('/')[-1]) for t in selected], + [(arm, n) for arm in E.ARMS for n in ('000', '004')]) + + def test_uncommitted_last_is_not_selected(self): + tasks = [task('paired', n) for n in range(4)] + ids = committed(tasks) + ids.remove(E.task_key(tasks[-1]) + '--verify') + selected, counts = E.select_tasks(tasks, ids) + self.assertEqual(selected, [tasks[0], tasks[2]]) + self.assertEqual(counts['paired'], 3) + + def test_probe_and_duplicate_selection(self): + train, probe = task('paired', 0), task('paired', 1, 'probe') + selected, _ = E.select_tasks([train, probe], committed([train, probe])) + self.assertEqual(selected, [train]) + with self.assertRaisesRegex(ValueError, 'Duplicate'): + E.select_tasks([train, train], committed([train])) + + def test_no_complete_triple_fails(self): + train = task('paired', 0) + with self.assertRaisesRegex(ValueError, 'No committed'): + E.select_tasks([train], {E.task_key(train) + '--verify'}) + + +class FileSafetyTests(unittest.TestCase): + def test_output_must_be_new_and_outside_protected(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary).resolve() + frozen = root / 'frozen' + frozen.mkdir() + with self.assertRaisesRegex(ValueError, 'fresh'): + E.validate_destination(frozen, [frozen]) + with self.assertRaisesRegex(ValueError, 'overlaps'): + E.validate_destination(frozen / 'new', [frozen]) + self.assertEqual(E.validate_destination(root / 'allowed', [frozen]), root / 'allowed') + self.assertFalse((root / 'allowed').exists()) + + def test_exclusive_save_and_mismatch_rejection(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary).resolve() / 'new.json' + E.save(path, {'a': 1}) + original = path.read_bytes() + with self.assertRaises(FileExistsError): + E.save(path, {'a': 2}) + self.assertEqual(path.read_bytes(), original) + with self.assertRaisesRegex(ValueError, 'Pinned bytes'): + E.check(path, '0' * 64) + + def test_duplicate_json_rejected(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary).resolve() / 'bad.json' + path.write_text('{"a":1,"a":2}', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'Duplicate'): + E.read(path) + + +class ProcessAuthorityTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name).resolve() + self.name = 'task-synthetic--verify' + self.job_path = self.root / 'jobs' / (self.name + '.json') + self.process_path = self.root / 'receipts' / self.name / 'process.json' + self.job_path.parent.mkdir() + self.process_path.parent.mkdir(parents=True) + E.save(self.job_path, dict(registration_sha256=E.REG_SHA, input_sha256={})) + self.process = dict(entry_id=self.name, registration_sha256=E.REG_SHA, + job_sha256=E.digest(self.job_path), complete=True, exit_code=0, timed_out=False, + exit_observed_by_parent=True, launch_error=None, argv=[str(E.ORIGINAL_ENV / 'bin/python'), '-B', 'validator.py']) + + def tearDown(self): + self.temp.cleanup() + + def verify(self, process): + E.save(self.process_path, process) + tracked = {str(p): E.identity(p) for p in (self.job_path, self.process_path)} + row = dict(entry_id=self.name, metadata=dict(job_file=str(self.job_path), + process_receipt_file=str(self.process_path), observed_returncode=0), consumed_inputs=[], + new_outputs=[dict(path=p, **info) for p, info in tracked.items()]) + start = dict(job=dict(path=str(self.job_path), sha256=E.digest(self.job_path))) + return E.validate_process(dict(row=row, start=start), tracked, self.root) + + def test_observed_complete_zero_is_required(self): + self.verify(self.process) + + def test_failed_or_unobserved_process_rejected(self): + self.process['exit_code'] = 1 + with self.assertRaisesRegex(ValueError, 'observed exit zero'): + self.verify(self.process) + + def test_boolean_zero_is_not_os_exit(self): + self.process['exit_code'] = False + with self.assertRaisesRegex(ValueError, 'observed exit zero'): + self.verify(self.process) + + def test_different_registration_rejected(self): + self.process['registration_sha256'] = '0' * 64 + with self.assertRaisesRegex(ValueError, 'observed exit zero'): + self.verify(self.process) + + def test_different_interpreter_rejected(self): + self.process['argv'][0] = '/different/venv/bin/python' + with self.assertRaisesRegex(ValueError, 'original -B'): + self.verify(self.process) + + +if __name__ == '__main__': + unittest.main() diff --git a/models/male-cns-lif/conditioning-acceleration-v1/test_gpu_math.py b/models/male-cns-lif/conditioning-acceleration-v1/test_gpu_math.py new file mode 100644 index 0000000..24a44ce --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/test_gpu_math.py @@ -0,0 +1,233 @@ +"""Host-only tests; CUDA equality requires separate real-device evidence.""" +from __future__ import annotations + +from dataclasses import replace +import importlib.util +import math +import os +from pathlib import Path +import sys +import unittest + +import numpy as np + +from gpu_backend import (CUDA_OPTIONS, CUDA_SOURCE, COMPARTMENTS, CudaEligibility, cpu_reference, + prepare_events) + + +ROOT = Path(__file__).resolve().parents[1] +PHASE4J = Path(os.environ.get("PHASE4J_SOURCE", str(ROOT / "models" / "male-cns-lif" / "phase4j.py"))) +BRIDGE_PATH = Path(os.environ.get("ELIGIBILITY_BRIDGE_SOURCE", str(ROOT / "wholebrain-plastic-interface-v1" / "eligibility_bridge.py"))) +spec = importlib.util.spec_from_file_location("_gpu_math_frozen_bridge", BRIDGE_PATH) +bridge = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = bridge +spec.loader.exec_module(bridge) + + +def assert_bits(test, left, right): + test.assertEqual(left.dtype, right.dtype) + test.assertEqual(left.shape, right.shape) + test.assertEqual(left.tobytes(), right.tobytes()) + + +class HostMathTests(unittest.TestCase): + def setUp(self): + self.kc = np.array([2, 5, 9], np.int32) + self.step = np.array([0, 0, 1, 2, 2, 5, 7], np.int32) + self.neurons = np.array([2, 4, 2, 5, 9, 9, 2], np.int32) + self.sources = np.array([2, 2, 5, 9], np.int32) + self.compartments = np.array([COMPARTMENTS[0], COMPARTMENTS[1], + COMPARTMENTS[0], COMPARTMENTS[0]]) + self.base = np.array([0.001, 0.01, 0.004, 0.08], np.float32) + self.delta = np.zeros(4, np.float32) + self.kw = dict(d=1.0, frozen=False, eta_mv=bridge.ETA_MV, + floor_fraction=bridge.FLOOR_FRACTION, + decay=math.exp(-bridge.DT_MS / bridge.TAU_MS)) + + def prepared(self, commands=(0, 2, 6), steps=8): + return prepare_events(self.step, self.neurons, self.kc, steps, commands, n_neurons=12) + + def run_host(self, prepared=None, **overrides): + args = self.kw | overrides + return cpu_reference(self.prepared() if prepared is None else prepared, + self.sources, self.compartments, COMPARTMENTS[0], + self.delta, self.base, **args) + + def test_matches_frozen_trace_and_update_and_no_input_mutation(self): + arrays = (self.step, self.neurons, self.kc, self.sources, + self.compartments, self.base, self.delta) + before = [(a.tobytes(), a.flags.writeable) for a in arrays] + prepared = self.prepared() + result = self.run_host(prepared) + trace = bridge.trace_from_events(self.step, self.neurons, self.kc, 8, [0, 2, 6], + phase4j_source=PHASE4J, n_neurons=12) + expected, _ = bridge.apply_eligibility(trace.last_command_trace, self.kc, + self.sources, self.compartments, COMPARTMENTS[0], self.delta, self.base, + d=1.0, frozen=False, phase4j_source=PHASE4J, n_neurons=12) + for left, right in ((result.snapshots, trace.snapshots), + (result.final_trace, trace.final), + (result.sampled_trace, trace.last_command_trace), + (result.delta_after, expected)): + assert_bits(self, left, right) + self.assertFalse(left.flags.writeable) + self.assertEqual(before, [(a.tobytes(), a.flags.writeable) for a in arrays]) + self.assertEqual(prepared.kc_event_count, 6) + self.assertEqual(prepared.indptr.tolist(), [0, 3, 4, 6]) + self.assertEqual(prepared.ticks.tolist(), [0, 1, 7, 2, 2, 5]) + + def test_no_command_returns_positive_zero_but_nonzero_final(self): + result = self.run_host(self.prepared(()), d=0) + self.assertEqual(result.snapshots.shape, (0, 3)) + assert_bits(self, result.sampled_trace, np.zeros(3, np.float32)) + self.assertTrue(np.any(result.final_trace > 0)) + assert_bits(self, result.delta_after, self.delta) + + def test_sampling_after_spike_and_last_command_only(self): + result = self.run_host(self.prepared((0, 2, 7))) + self.assertEqual(result.snapshots[0, 0], np.float32(1)) + self.assertEqual(result.snapshots[1, 1], np.float32(1)) + assert_bits(self, result.sampled_trace, result.snapshots[-1]) + assert_bits(self, result.final_trace, result.snapshots[-1]) + + def test_frozen_d_zero_off_compartment_and_floor(self): + self.delta = np.array([-0.0001, -0.0005, -0.001, 0], np.float32) + for kwargs in (dict(d=0), dict(frozen=True)): + assert_bits(self, self.run_host(**kwargs).delta_after, self.delta) + result = self.run_host() + assert_bits(self, result.delta_after[1:2], self.delta[1:2]) + self.assertEqual(result.delta_after[0], np.float32(-0.25) * self.base[0]) + self.assertEqual(result.delta_after[2], self.delta[2]) + + def test_empty_events_and_one_tick(self): + prepared = prepare_events(np.empty(0, np.int32), np.empty(0, np.int32), + self.kc, 1, [0], n_neurons=12) + result = self.run_host(prepared) + assert_bits(self, result.final_trace, np.zeros(3, np.float32)) + assert_bits(self, result.delta_after, self.delta) + + def test_long_empty_gap_does_not_use_power(self): + self.step = np.array([0], np.int32) + self.neurons = np.array([2], np.int32) + prepared = self.prepared((9999,), steps=10000) + result = self.run_host(prepared) + expected = np.float32(1) + for _ in range(9999): + expected = np.float32(expected * np.float32(self.kw["decay"])) + assert_bits(self, result.sampled_trace[:1], np.array([expected], np.float32)) + jumped = np.float32(float(np.float32(self.kw["decay"])) ** 9999) + self.assertNotEqual(expected.view(np.uint32), jumped.view(np.uint32)) + + def test_random_sparse_case_matches_frozen(self): + rng = np.random.default_rng(717) + self.kc = np.arange(0, 80, 2, dtype=np.int32) + encoded = np.unique(rng.integers(0, 80 * 257, size=1400)) + self.step = (encoded // 80).astype(np.int32) + self.neurons = (encoded % 80).astype(np.int32) + self.sources = rng.choice(self.kc, 159).astype(np.int32) + self.compartments = rng.choice(np.asarray(COMPARTMENTS), 159) + self.base = rng.uniform(0.0001, 2, 159).astype(np.float32) + self.delta = -(rng.uniform(0, 0.24, 159).astype(np.float32) * self.base) + prepared = prepare_events(self.step, self.neurons, self.kc, 257, [0, 1, 20, 99, 256], n_neurons=80) + result = self.run_host(prepared) + trace = bridge.trace_from_events(self.step, self.neurons, self.kc, 257, + [0, 1, 20, 99, 256], phase4j_source=PHASE4J, n_neurons=80) + expected, _ = bridge.apply_eligibility(trace.last_command_trace, self.kc, + self.sources, self.compartments, COMPARTMENTS[0], self.delta, self.base, + d=1, phase4j_source=PHASE4J, n_neurons=80) + assert_bits(self, result.snapshots, trace.snapshots) + assert_bits(self, result.delta_after, expected) + + def test_signed_zero_source_floor_tie_is_preserved(self): + # The bridge's identity guard rejects this pathological subnormal base. + # Compare the original apply_v2 directly: np.maximum(+0,-0) gives -0. + self.base[:] = np.nextafter(np.float32(0), np.float32(1)) + result = self.run_host(d=0) + original, _, _ = bridge._load_original(PHASE4J) + from types import SimpleNamespace + validator = SimpleNamespace(n=12, kc=self.kc, base_candidate=self.base, + cand={"source_neuron_idx": bridge._Column(self.sources), + "target_compartment_verbatim": bridge._Column(self.compartments)}) + expected, _ = original(validator, result.sampled_trace, COMPARTMENTS[0], 0, + self.delta, {"eligibility_v2": {"eta_effective_mv": bridge.ETA_MV, + "max_fractional_depression": bridge.FLOOR_FRACTION}}) + assert_bits(self, result.delta_after, expected) + self.assertTrue(np.all(np.signbit(result.delta_after))) + assert_bits(self, self.run_host(frozen=True).delta_after, self.delta) + + def test_input_validation(self): + bad_events = [ + (self.step.astype(np.int64), self.neurons, self.kc, 8, [0]), + (self.step, self.neurons.astype(np.float32), self.kc, 8, [0]), + (self.step, self.neurons, self.kc[::-1], 8, [0]), + (self.step, self.neurons, self.kc, 8, [0, 0]), + (self.step, self.neurons, self.kc, 8, [0.0]), + (self.step, self.neurons, self.kc, 8, [8]), + (self.step, self.neurons, self.kc, True, [0]), + (self.step, self.neurons, self.kc, 0, []), + (np.array([0, 0], np.int32), np.array([2, 2], np.int32), self.kc, 8, []), + ] + for args in bad_events: + with self.subTest(args=args), self.assertRaises(ValueError): + prepare_events(*args, n_neurons=12) + for kwargs in (dict(d=True), dict(d=0.5), dict(d=float("nan")), + dict(frozen=1), dict(eta_mv=float("inf")), + dict(floor_fraction=0), dict(decay=float("nan"))): + with self.subTest(kwargs=kwargs), self.assertRaises(ValueError): + self.run_host(**kwargs) + self.delta[0] = -0.0 + with self.assertRaises(ValueError): + self.run_host() + self.delta[0] = 0 + self.base[0] = np.nan + with self.assertRaises(ValueError): + self.run_host() + + def test_forged_prepared_rejected(self): + prepared = self.prepared() + bad = replace(prepared, ticks=np.array([0, 0, 7, 2, 2, 5], np.int32)) + with self.assertRaises(ValueError): + self.run_host(bad) + + def test_cuda_entry_rejects_bad_inputs_before_accessing_device(self): + # An uninitialized shell has no cp/device fields. Rejection therefore + # proves the CUDA entry's validation happens before any CUDA operation. + backend = object.__new__(CudaEligibility) + prepared = self.prepared() + base_args = dict(prepared=prepared, candidate_sources=self.sources, + candidate_compartments=self.compartments, compartment=COMPARTMENTS[0], + delta_before=self.delta, base_candidate=self.base, **self.kw) + bad = [dict(candidate_sources=self.sources.astype(np.int64)), + dict(candidate_compartments=self.compartments.astype("S")), + dict(delta_before=self.delta.astype(np.float64)), + dict(base_candidate=np.array([np.inf, 1, 1, 1], np.float32)), + dict(delta_before=np.array([np.nan, 0, 0, 0], np.float32)), + dict(candidate_sources=np.array([1, 2, 5, 9], np.int32)), + dict(candidate_sources=np.array([2, 2, 5, 11], np.int32)), + dict(delta_before=np.array([-1, 0, 0, 0], np.float32)), + dict(compartment="unknown"), dict(kernel_warmup=-1), + dict(kernel_repeats=0), dict(kernel_repeats=True)] + for overrides in bad: + with self.subTest(overrides=overrides), self.assertRaises(ValueError): + backend.run(**(base_args | overrides)) + + def test_no_dense_tick_raster_and_input_storage_owned(self): + prepared = self.prepared() + for arr in (prepared.kc, prepared.indptr, prepared.ticks, prepared.command_steps): + self.assertEqual(arr.ndim, 1) + self.assertFalse(arr.flags.writeable) + self.kc[:] = 0 + self.step[:] = 0 + self.assertEqual(prepared.kc.tolist(), [2, 5, 9]) + self.assertEqual(prepared.ticks.tolist(), [0, 1, 7, 2, 2, 5]) + + def test_kernel_explicit_rounding_controls(self): + self.assertIn("--fmad=false", CUDA_OPTIONS) + self.assertIn("--ftz=false", CUDA_OPTIONS) + self.assertIn("__fmul_rn", CUDA_SOURCE) + self.assertIn("__fadd_rn", CUDA_SOURCE) + self.assertNotIn("pow(", CUDA_SOURCE) + self.assertNotIn("__expf", CUDA_SOURCE) + + +if __name__ == "__main__": + unittest.main()