From c66cbcd2237c4bec6b688641b324b5c6802eea3d Mon Sep 17 00:00:00 2001 From: Digital Drosophila Archive Date: Thu, 1 Oct 2026 15:06:13 +0800 Subject: [PATCH] Audit unchanged full-brain fresh and persistent replay equivalence --- .../brain_replay.py | 566 ++++++++++++++++++ .../test_brain_replay.py | 161 +++++ 2 files changed, 727 insertions(+) create mode 100644 models/male-cns-lif/conditioning-pipeline-acceleration-v1/brain_replay.py create mode 100644 models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_brain_replay.py diff --git a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/brain_replay.py b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/brain_replay.py new file mode 100644 index 0000000..f930583 --- /dev/null +++ b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/brain_replay.py @@ -0,0 +1,566 @@ +"""Bounded unchanged-runtime replay: 14 fresh execs versus one resident exec. + +Only the original R.run_light is executed, in the original 4090 environment. +Every call allocates fresh neural/APL state. No GPU graph residency, numerical +optimization, conditioning coordinator, generated overlay, or AR resume is used. +All output is new, outside the scientific tree. Original checkpoints are read. +""" +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 signal +import subprocess +import sys +import time +import traceback + + +ROOT = Path('/data/digital-drosophila') +MODEL = ROOT / 'models/male-cns-lif' +EXP = ROOT / 'experiments/phase4ar-normal-conditioning' +ORIGINAL_ENV = MODEL / '.venv' +ORIGINAL_PYTHON = ORIGINAL_ENV / 'bin/python' +REG_PATH = EXP / 'protocol/preregistered.json' +REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740' +SOURCE_COMMIT = '80aeb2093db655329774eff7b32fb6e417b8b40b' +GPU = 'GPU-79f3a85d-adce-7833-7003-6488140ca596' +R_PATH = MODEL / 'olfactory-conditioning-lab-v1/runtime_light.py' +C_PATH = MODEL / 'olfactory-conditioning-lab-v1/runtime_continuation.py' +AP_PATH = MODEL / 'conditioning-interface-v2/isolated_workers.py' +R_SHA = '522b8c31ea938890fe15364fccd25ccc5bded54101754909f255532e7ad01aae' +C_SHA = '0b8e8e0a6a23a079dc62a88355ce355d53b3ae22f6665519a8f6aef2dcfc5aba' +AP_SHA = 'e47651a8ea99d7c44115107ff6b5dc71312183609ae11c891db65bb21ed61e30' +STATE_KEYS = ('step', 'neuron_idx', 'counts', 'v', 'g', 'ref') +TERMINAL_KEYS = ('spike_ring', 'apl_state', 'apl_pending', 'next_tick') +ARMS = ('cs_only', 'd_zero', 'dan_only', 'frozen', 'paired', 'random_reward', 'unpaired') +SUMMARY_KEYS = ('completed_ticks', 'nodes', 'edges', 'precision', 'apl_precision', 'state_units', + 'total_spikes', 'requested', 'accepted_requests', 'rejected_requests', 'arriving_edges', + 'accepted_arriving_edges', 'counts_match_sparse_events', 'all_finite', 'record_history', + 'apl_fast_edges_zeroed', 'kc_targets_present', 'effective_weights_raw_sha256', 'apl_schedule', + 'apl_si_conversion', 'apl_delay_steps', 'pending_generated_count', 'pending_initial_zero_prefix', + 'pending_tail_shape', 'future_generated_decay_tail_included', 'diagnostic_mode', + 'within_trial_weight_updates', 'checkpoint_continuation_used', 'explicit_steps_executed', + 'first_failure_tick') + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def path_checked(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 sha(path): + digest = hashlib.sha256() + with path_checked(path).open('rb') as stream: + for block in iter(lambda: stream.read(8 << 20), b''): + digest.update(block) + return digest.hexdigest() + + +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(path_checked(path).read_text(encoding='utf-8'), object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(ValueError('Nonfinite JSON'))) + + +def save(path, value): + with path_checked(path).open('x', encoding='utf-8', newline='\n') as stream: + json.dump(value, stream, indent=2, sort_keys=True, allow_nan=False) + stream.write('\n') + + +def pin_add(pins, path, expected): + path = str(path_checked(path)) + expected = expected if isinstance(expected, str) else expected['sha256'] + require(isinstance(expected, str) and len(expected) == 64 + and all(c in '0123456789abcdef' for c in expected), 'Invalid SHA256') + require(path not in pins or pins[path] == expected, 'Conflicting authority pin: ' + path) + pins[path] = expected + + +def verify(pins): + for path, expected in pins.items(): + require(sha(path) == expected, 'Pinned input changed: ' + path) + + +def contained(root, relative): + path = path_checked(root / relative) + require(path.is_relative_to(root), 'Fixture path leaves exported root') + return path + + +def load_source(name, path, expected): + raw = path_checked(path).read_bytes() + require(hashlib.sha256(raw).hexdigest() == expected and name not in sys.modules, + 'Unverified or repeated module binding: ' + name) + 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 fixture_order(count): + require(count == 14, 'The fixed manifest must contain exactly 14 real fixtures') + return [('forward', index) for index in range(count)] + [('reverse', index) for index in range(count - 1, -1, -1)] + + +def build_selection(fixtures, manifest_sha): + """Bind saved case/overlay and old checkpoint to the authenticated export.""" + fixtures = path_checked(fixtures) + manifest_path = fixtures / 'manifest.json' + require(sha(manifest_path) == manifest_sha, 'Export manifest differs from external pin') + manifest = read(manifest_path) + require(manifest.get('complete') is True and manifest.get('schema') == 'conditioning-acceleration-fixtures-v1' + and manifest.get('registration_sha256') == REG_SHA and manifest.get('source_code_commit') == SOURCE_COMMIT, + 'Unexpected completed real export authority') + records = manifest['fixtures'] + fixture_order(len(records)) + require(len({record['fixture_id'] for record in records}) == 14, 'Duplicate fixture identity') + require({arm: sum(row['task']['arm'] == arm for row in records) for arm in ARMS} + == {arm: 2 for arm in ARMS}, 'Expected the declared two fixtures per available arm') + require(sha(REG_PATH) == REG_SHA, 'Original AR registration changed') + registration = read(REG_PATH) + require(registration['source_code_commit'] == SOURCE_COMMIT + and registration['scientific_scope'] == 'normal-acquisition-only', 'Original registration role differs') + global_pins = {str(manifest_path): manifest_sha, str(REG_PATH): REG_SHA, + str(R_PATH): R_SHA, str(C_PATH): C_SHA, str(AP_PATH): AP_SHA} + for relative, info in manifest['files'].items(): + pin_add(global_pins, contained(fixtures, relative), info) + verify(global_pins) + rows = [] + for record in records: + name = record['fixture_id'] + directory = contained(fixtures, name) + require(read(directory / 'fixture.json') == record + 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, + 'Fixture lacks original full mapping/reference evidence') + pins = dict(global_pins) + for path, info in read(directory / 'source-file-pins.json').items(): + pin_add(pins, path, info) + task = record['task'] + key = 'task-' + hashlib.sha256(task['task_id'].encode()).hexdigest() + brain_name = key + '--brain' + job_path = EXP / 'jobs' / (brain_name + '.json') + process_path = EXP / 'receipts' / brain_name / 'process.json' + authority_path = EXP / 'jobs' / (key + '.authority.json') + require(all(str(p) in pins for p in (job_path, process_path, authority_path)), + 'Original brain job/process/authority not bound by exported source pins') + verify({str(p): pins[str(p)] for p in (job_path, process_path, authority_path)}) + job, process, authority = read(job_path), read(process_path), read(authority_path) + require(authority == read(directory / 'authority.json') + and sha(authority_path) == record['authority_sha256'], 'Export authority differs from sealed original') + require(job['kind'] == 'brain' and job['job_id'] == brain_name + and job['schema'] == 'phase4ap-isolated-worker-job-v1' + and job['registration_sha256'] == REG_SHA and job['worker_sha256'] == AP_SHA + and job['diagnostic_mode'] == 'light', 'Original brain job differs') + require(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 process['job_sha256'] == pins[str(job_path)] + and process['entry_id'] == brain_name and process['registration_sha256'] == REG_SHA + and process['argv'][0] == str(ORIGINAL_PYTHON) and '-B' in process['argv'], + 'Original brain has no authenticated successful fresh exec') + require(job['sources'] == {'runtime_light': {'path': str(R_PATH), 'sha256': R_SHA}, + 'runtime_continuation': {'path': str(C_PATH), 'sha256': C_SHA}}, + 'Original brain sources differ') + for path, expected in job['input_sha256'].items(): + pin_add(pins, path, expected) + for path, expected in authority['frozen_inputs'].items(): + pin_add(pins, path, expected) + for info in authority['files'].values(): + pin_add(pins, info['path'], info['sha256']) + case = path_checked(job['case_dir']) + old = path_checked(job['output_dir']) + require(case == EXP / 'cases' / key and old == EXP / 'runs' / key + and authority['checkpoint_dir'] == str(old) and authority['epoch_id'] == task['task_id'], + 'Old brain case/checkpoint/task binding differs') + for role, path in (('spec', case / 'spec.json'), ('requests', case / 'requests.npz'), + ('overlay_before', Path(job['overlay_file'])), + ('overlay_manifest_before', Path(job['overlay_manifest']))): + require(authority['files'][role] == {'path': str(path), 'sha256': pins[str(path)]}, + 'Original brain input role differs: ' + role) + spec = read(case / 'spec.json') + require(spec['steps'] == spec['max_steps'] == 1200 and spec['record_history'] is False + and spec['apl_delay_steps'] == 18 and spec['dt_ms'] == .1 + and spec['within_trial_weight_updates'] is False, 'Replay is not the exact bounded fresh trial') + require(not (case / 'arrays.npz').exists(), 'Unexpected neural initial-state override') + old_manifest = old / 'checkpoint-manifest.json' + pin_add(pins, old_manifest, authority['checkpoint_manifest_sha256']) + require(sha(old_manifest) == authority['checkpoint_manifest_sha256'], 'Old checkpoint manifest changed') + checkpoint = read(old_manifest) + require(checkpoint['schema'] == 'phase4an-checkpoint-v1' and checkpoint['complete'] is True + and checkpoint['segment_steps'] == 1200 and checkpoint['global_tick_offset'] == 0 + and checkpoint['global_next_tick'] == 1200, 'Expected sealed fresh reference checkpoint') + for relative, expected in checkpoint['files_sha256'].items(): + pin_add(pins, contained(old, relative), expected) + require(checkpoint['files_sha256']['state.npz'] == authority['files']['state']['sha256'], + 'Authority state differs from complete checkpoint inventory') + rows.append(dict(fixture_id=name, task_id=task['task_id'], case_dir=str(case), + overlay_file=job['overlay_file'], overlay_manifest=job['overlay_manifest'], + original_checkpoint=str(old), original_checkpoint_manifest_sha256=authority['checkpoint_manifest_sha256'], + original_brain_job=str(job_path), original_process_wall_seconds=process['end_to_end_seconds'], + pins=pins)) + # All original and exported bytes are verified before any new GPU work. + all_pins = dict(global_pins) + for row in rows: + for path, expected in row['pins'].items(): + pin_add(all_pins, path, expected) + verify(all_pins) + return dict(schema='conditioning-persistent-process-selection-v1', registration_sha256=REG_SHA, + fixture_manifest=str(manifest_path), fixture_manifest_sha256=manifest_sha, + fixed_count=14, fresh_order=list(range(14)), resident_order=fixture_order(14), + selected_before_candidate_execution=True, fixtures=rows, all_pins=all_pins, + numerical_implementation='unchanged frozen runtime_light.run_light', + neural_reset='original function allocates fresh neuron/APL/queues for every call', + request_rng='none during runtime; exact saved requests.npz reused', + no_GPU_graph_residency=True, no_generated_overlay_installed=True) + + +def array_evidence(array): + return dict(dtype=array.dtype.str, shape=list(array.shape), + sha256=hashlib.sha256(array.tobytes(order='C')).hexdigest()) + + +def compare_arrays(actual, expected, required): + require(set(actual) == set(expected) == set(required), 'Checkpoint array schema differs') + result = {} + for name in required: + left, right = array_evidence(actual[name]), array_evidence(expected[name]) + result[name] = dict(bitwise_exact=left == right, actual=left, expected=right) + return result + + +def compare_checkpoints(current, original, current_dir, original_dir): + arrays = dict(state=compare_arrays(current['state'], original['state'], STATE_KEYS), + terminal=compare_arrays(current['terminal'], original['terminal'], TERMINAL_KEYS)) + summaries = {key: dict(actual=current['summary'].get(key), expected=original['summary'].get(key), + exact=key in current['summary'] and key in original['summary'] + and current['summary'][key] == original['summary'][key]) for key in SUMMARY_KEYS} + files = {name: dict(actual=sha(current_dir / name), expected=sha(original_dir / name)) + for name in ('requests.csv', 'finite-audit.json', 'apl-pending-tail.float32.npy')} + for item in files.values(): + item['exact'] = item['actual'] == item['expected'] + environment_equal = current['summary']['apl_component']['environment'] == original['summary']['apl_component']['environment'] + exact = (all(item['bitwise_exact'] for group in arrays.values() for item in group.values()) + and all(item['exact'] for item in summaries.values()) + and all(item['exact'] for item in files.values()) and environment_equal + and current['fresh'] is original['fresh'] is True + and current['global_tick_offset'] == original['global_tick_offset'] == 0) + return dict(exact=exact, arrays=arrays, summaries=summaries, files=files, + original_APL_environment_exact=environment_equal, + comparison_scope='all six state arrays, four terminal arrays, requests, finite coverage, APL pending, effective weights and selected summaries') + + +def check_environment(): + require(sys.dont_write_bytecode and Path(sys.prefix) == ORIGINAL_ENV, + 'Use the original registered environment Python -B') + require(os.environ.get('CUDA_VISIBLE_DEVICES') == GPU + and os.environ.get('CUDA_HOME') == '/usr/local/cuda' + and os.environ.get('TORCH_CUDA_ARCH_LIST') == '8.9', 'Explicit original 4090 build environment required') + + +def run_worker(job_path, job_sha): + """One exec; resident lane shares only original modules/caches, never state.""" + started = time.perf_counter() + check_environment() + require('torch' not in sys.modules and 'runtime_light' not in sys.modules, 'Worker must begin as a fresh exec') + require(sha(job_path) == job_sha, 'Replay worker job changed') + job = read(job_path) + require(sha(Path(__file__).absolute()) == job['candidate_source_sha256'], 'Candidate source changed') + require(sha(job['selection_file']) == job['selection_sha256'], 'Pinned selection changed') + selection = read(job['selection_file']) + require(selection['registration_sha256'] == REG_SHA and selection['fixed_count'] == 14, 'Worker selection authority differs') + count = 1 if job['lane'] == 'fresh' else 28 + require(job['lane'] in ('fresh', 'resident') and len(job['runs']) == count, 'Worker lane count differs') + if job['lane'] == 'resident': + require(job['runs'] == [[direction, index] for direction, index in fixture_order(14)], 'Resident order differs') + output = path_checked(job['output_root']) + require(output.is_dir() and output.parent == path_checked(job['experiment_root']), 'Worker output root differs') + load_started = time.perf_counter() + ap = load_source('_pipeline_original_AP', AP_PATH, AP_SHA) + runtime = load_source('runtime_light', R_PATH, R_SHA) + checkpoint = None + initial_module_seconds = time.perf_counter() - load_started + records = [] + for sequence, (direction, index) in enumerate(job['runs']): + require(time.time() < job['deadline_unix'], 'Bounded replay wall deadline reached') + require(type(index) is int and 0 <= index < 14, 'Invalid fixed fixture index') + row = selection['fixtures'][index] + trial_started = time.perf_counter() + authority_started = time.perf_counter() + verify(row['pins']) + require(sha(job_path) == job_sha and sha(job['selection_file']) == job['selection_sha256'], 'Worker authority changed') + admission = ap._admit_brain() + authority_before_seconds = time.perf_counter() - authority_started + name = (f'{index:02d}-' if job['lane'] == 'fresh' else f'{sequence:02d}-{direction}-') + row['fixture_id'] + trial_root = output / name + trial_root.mkdir(exist_ok=False) + trial_dir = trial_root / 'trial' + runtime_started = time.perf_counter() + summary = runtime.run_light(Path(row['case_dir']), trial_dir, Path(job['build_root']), + Path(row['overlay_file']), Path(row['overlay_manifest'])) + runtime_seconds = time.perf_counter() - runtime_started + checkpoint_started = time.perf_counter() + if checkpoint is None: + checkpoint = load_source('_pipeline_original_checkpoint', C_PATH, C_SHA) + seal = checkpoint.seal_checkpoint(trial_dir) + current = checkpoint.load_checkpoint(trial_dir, seal['sha256']) + original = checkpoint.load_checkpoint(row['original_checkpoint'], row['original_checkpoint_manifest_sha256']) + fresh_reference = None + if job['lane'] == 'resident': + fresh_pin = job['fresh_references'][row['fixture_id']] + require(sha(fresh_pin['report']) == fresh_pin['report_sha256'], 'Fresh reference report changed') + fresh_reference = checkpoint.load_checkpoint(fresh_pin['checkpoint'], fresh_pin['checkpoint_manifest_sha256']) + require(current['summary'] == summary, 'Current sealed summary differs from actual returned summary') + checkpoint_seconds = time.perf_counter() - checkpoint_started + compare_started = time.perf_counter() + environment = ap._environment('brain') + ap._verify_precision(environment, summary['apl_component']['environment']) + comparison = compare_checkpoints(current, original, trial_dir, Path(row['original_checkpoint'])) + if fresh_reference is not None: + comparison['fresh_reference'] = compare_checkpoints(current, fresh_reference, trial_dir, Path(fresh_pin['checkpoint'])) + comparison['exact'] = comparison['exact'] and comparison['fresh_reference']['exact'] + compare_seconds = time.perf_counter() - compare_started + io_started = time.perf_counter() + save(trial_root / 'comparison.json', comparison) + comparison_io_seconds = time.perf_counter() - io_started + require(comparison['exact'], 'Brain replay differs from original sealed checkpoint: ' + row['fixture_id']) + after_started = time.perf_counter() + verify(row['pins']) + require(sha(Path(__file__).absolute()) == job['candidate_source_sha256'], 'Candidate source changed during trial') + require(sha(job_path) == job_sha and sha(job['selection_file']) == job['selection_sha256'], 'Worker authority changed after trial') + authority_after_seconds = time.perf_counter() - after_started + report = dict(fixture_id=row['fixture_id'], direction=direction, sequence=sequence, + fixed_index=index, lane=job['lane'], exact=True, original_checkpoint=row['original_checkpoint'], + original_checkpoint_manifest_sha256=row['original_checkpoint_manifest_sha256'], + current_checkpoint=str(trial_dir), current_checkpoint_manifest_sha256=seal['sha256'], + original_brain_job=row['original_brain_job'], input_pins_verified_before_and_after=True, + compared_array_count=len(STATE_KEYS) + len(TERMINAL_KEYS), comparison_sha256=sha(trial_root / 'comparison.json'), + admission=admission, environment=environment, + timings=dict(authority_before_seconds=authority_before_seconds, runtime_and_builtin_output_seconds=runtime_seconds, + checkpoint_seal_and_load_seconds=checkpoint_seconds, comparison_seconds=compare_seconds, + comparison_record_io_seconds=comparison_io_seconds, authority_after_seconds=authority_after_seconds, + original_build_or_cache_seconds=summary['build_or_cache_seconds'], + original_diagnostic_simulation_seconds=summary['diagnostic_simulation_seconds'], + trial_wall_before_report_seconds=time.perf_counter() - trial_started), + timing_limit='Original runtime includes its own I/O; it is not instrumented or subtracted. Build/import costs remain in recorded wall time.', + process_pid=os.getpid(), process_exit_claimed=False, + neural_state_fresh_each_call=True, within_trial_weight_updates=False, + only_saved_before_overlay_used=True, generated_delta_installed=False, original_experiment_resumed=False, + learning_success_claimed=False, GPU_graph_residency=False) + report_started = time.perf_counter() + save(trial_root / 'replay-report.json', report) + records.append(dict(report=str(trial_root / 'replay-report.json'), sha256=sha(trial_root / 'replay-report.json'), + report_io_seconds=time.perf_counter() - report_started, + trial_wall_seconds=time.perf_counter() - trial_started, **report)) + print(json.dumps(dict(lane=job['lane'], fixture=row['fixture_id'], direction=direction, exact=True)), flush=True) + # Checkpoints hold owned CPU arrays only; no state is fed to the next run. + del current, original, fresh_reference + verify(selection['all_pins']) + result = dict(complete=True, lane=job['lane'], count=len(records), records=records, + process_pid=os.getpid(), initial_module_load_seconds=initial_module_seconds, + worker_wall_seconds=time.perf_counter() - started, all_bitwise_exact=True, + original_runtime_unchanged=True, fresh_neural_state_every_trial=True, + process_exit_independently_observed=False, original_experiment_resumed=False, + original_files_modified=False, scientific_learning_pass_claimed=False) + save(Path(job['result_file']), result) + return result + + +def launch(job, job_path, process_dir, deadline): + save(job_path, job) + job_sha = sha(job_path) + argv = [str(ORIGINAL_PYTHON), '-B', str(Path(__file__).absolute()), + '--worker-job', str(job_path), '--worker-job-sha256', job_sha] + environment = dict(os.environ, CUDA_VISIBLE_DEVICES=GPU, CUDA_HOME='/usr/local/cuda', + MAX_JOBS='2', TORCH_CUDA_ARCH_LIST='8.9', PYTHONDONTWRITEBYTECODE='1', + PATH=str(ORIGINAL_ENV / 'bin') + os.pathsep + '/usr/local/cuda/bin' + os.pathsep + os.environ.get('PATH', '')) + label = job_path.stem + stdout_path, stderr_path = process_dir / (label + '.stdout.log'), process_dir / (label + '.stderr.log') + started = time.perf_counter() + timeout = deadline - time.time() + require(timeout > 0, 'Replay wall budget exhausted before spawn') + returncode, timed_out, launch_error = None, False, None + process = None + terminated_child = False + + def stop_child(): + nonlocal terminated_child + if process is None: + return + terminated_child = True + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + + def interrupted(signum, _frame): + stop_child() + raise KeyboardInterrupt('Replay interrupted by signal ' + str(signum)) + + previous_handlers = {signum: signal.getsignal(signum) for signum in (signal.SIGTERM, signal.SIGINT)} + try: + for signum in previous_handlers: + signal.signal(signum, interrupted) + with stdout_path.open('xb') as stdout, stderr_path.open('xb') as stderr: + process = subprocess.Popen(argv, cwd=Path(__file__).absolute().parent, env=environment, + stdout=stdout, stderr=stderr) + returncode = process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + stop_child() + returncode = process.returncode + except OSError as exc: + launch_error = f'{type(exc).__name__}: {exc}' + stop_child() + except BaseException: + stop_child() + raise + finally: + for signum, handler in previous_handlers.items(): + signal.signal(signum, handler) + receipt = dict(argv=argv, job_sha256=job_sha, actual_returncode=returncode, + timed_out=timed_out, launch_error=launch_error, child_whole_wall_seconds=time.perf_counter() - started, + exit_independently_observed=returncode is not None, parent_pid=os.getpid(), + child_pid=None if process is None else process.pid, + child_terminated_on_failure=terminated_child, + process_group_policy='inherits outer run_job owned process group; no nested session', + stdout_sha256=sha(stdout_path) if stdout_path.exists() else None, + stderr_sha256=sha(stderr_path) if stderr_path.exists() else None, + receipt_scope='actual exec process; resident trial reports never claim individual process exits') + save(process_dir / (label + '.json'), receipt) + require(returncode == 0 and not timed_out and launch_error is None, 'Replay child failed; evidence retained: ' + label) + result = read(job['result_file']) + require(result['complete'] is True and result['all_bitwise_exact'] is True + and result['lane'] == job['lane'] and result['count'] == len(job['runs']), 'Child result incomplete') + for row in result['records']: + require(sha(row['report']) == row['sha256'], 'Child trial report changed after exit') + receipt['result_file'] = job['result_file'] + receipt['result_sha256'] = sha(job['result_file']) + return receipt, result + + +def run(fixtures, manifest_sha, output): + started = time.perf_counter() + deadline = time.time() + 900 + require(sys.dont_write_bytecode and Path(sys.prefix) == ORIGINAL_ENV, + 'Parent must use the original registered environment Python -B') + output = path_checked(output) + require(not output.exists() and output.parent.is_dir() and not output.is_relative_to(ROOT) + and not output.is_relative_to(path_checked(fixtures)) + and not output.is_relative_to(Path(__file__).absolute().parent), 'Use a fresh output root outside all protected trees') + output.mkdir() + for name in ('fresh', 'resident', 'jobs', 'processes', 'build', 'sessions'): + (output / name).mkdir() + try: + preparation_started = time.perf_counter() + selection = build_selection(fixtures, manifest_sha) + selection_path = output / 'selection.json' + save(selection_path, selection) + candidate_sha = sha(Path(__file__).absolute()) + save(output / 'authority.json', dict(registration_sha256=REG_SHA, fixture_manifest_sha256=manifest_sha, + candidate_source_sha256=candidate_sha, selection_sha256=sha(selection_path), + original_interpreter=str(ORIGINAL_PYTHON), gpu_uuid=GPU, + fixed_trials=dict(fresh=14, resident_forward=14, resident_reverse=14), + wall_budget_seconds=900, deadline_unix=deadline, + preparation_seconds=time.perf_counter() - preparation_started, + source_pins={str(R_PATH): R_SHA, str(C_PATH): C_SHA, str(AP_PATH): AP_SHA}, + cold_build_policy='one new shared candidate build directory; first fresh call includes cold build, no warm-up trials', + existing_AR_not_resumed=True, numerical_implementation_unchanged=True, + created_utc=datetime.now(timezone.utc).isoformat())) + common = dict(selection_file=str(selection_path), selection_sha256=sha(selection_path), + candidate_source_sha256=candidate_sha, build_root=str(output / 'build'), + experiment_root=str(output), deadline_unix=deadline) + children, fresh_records = [], [] + for index in range(14): + label = f'fresh-{index:02d}' + job = dict(common, lane='fresh', runs=[['fresh', index]], output_root=str(output / 'fresh'), + result_file=str(output / 'sessions' / (label + '.json'))) + process, result = launch(job, output / 'jobs' / (label + '.json'), output / 'processes', deadline) + children.append(process) + fresh_records.extend(result['records']) + fresh_references = {row['fixture_id']: dict(checkpoint=row['current_checkpoint'], + checkpoint_manifest_sha256=row['current_checkpoint_manifest_sha256'], + report=row['report'], report_sha256=row['sha256']) for row in fresh_records} + job = dict(common, lane='resident', runs=fixture_order(14), output_root=str(output / 'resident'), + fresh_references=fresh_references, + result_file=str(output / 'sessions/resident.json')) + process, resident = launch(job, output / 'jobs/resident.json', output / 'processes', deadline) + children.append(process) + require(time.time() < deadline, 'Replay exceeded bounded wall budget') + verify(selection['all_pins']) + require(sha(Path(__file__).absolute()) == candidate_sha, 'Candidate source changed') + by_fresh = {row['fixture_id']: row for row in fresh_records} + paired = [] + for row in resident['records']: + reference = by_fresh[row['fixture_id']] + paired.append(dict(fixture_id=row['fixture_id'], direction=row['direction'], + fresh_trial_wall_seconds=reference['trial_wall_seconds'], + resident_trial_wall_seconds=row['trial_wall_seconds'], + fresh_runtime_and_output_seconds=reference['timings']['runtime_and_builtin_output_seconds'], + resident_runtime_and_output_seconds=row['timings']['runtime_and_builtin_output_seconds'], + both_exact_to_same_original_checkpoint=True)) + result = dict(status='PASS-UNCHANGED-R-PERSISTENT-PROCESS-REPLAY', complete=True, + fixture_count=14, fresh_trial_count=14, resident_trial_count=28, actual_child_processes=len(children), + all_bitwise_exact=True, children=children, matched_pairs=paired, + fresh_children_whole_wall_seconds=sum(c['child_whole_wall_seconds'] for c in children[:-1]), + resident_child_whole_wall_seconds=children[-1]['child_whole_wall_seconds'], + wall_seconds=time.perf_counter() - started, original_files_verified_unchanged=True, + source_runtime_unchanged=True, neural_state_fresh_every_call=True, GPU_graph_residency=False, + original_experiment_resumed=False, generated_overlay_installed=False, + scientific_learning_pass_claimed=False, + timing_limit='Different lane counts (14 vs 28); use per-fixture pairs. First fresh includes cold extension build. No pure kernel speedup is claimed.', + files={str(p.relative_to(output)): sha(p) for p in sorted(output.rglob('*')) if p.is_file()}) + save(output / 'summary.json', result) + return result + except BaseException: + save(output / 'failure.json', dict(complete=False, traceback=traceback.format_exc(), + wall_seconds=time.perf_counter() - started, original_experiment_resumed=False, + automatic_retry=False, partial_outputs_retained=True)) + raise + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--fixtures', type=Path) + parser.add_argument('--fixture-manifest-sha256') + parser.add_argument('--output', type=Path) + parser.add_argument('--worker-job', type=Path, help=argparse.SUPPRESS) + parser.add_argument('--worker-job-sha256', help=argparse.SUPPRESS) + args = parser.parse_args() + if args.worker_job: + require(args.worker_job_sha256 and not any((args.fixtures, args.fixture_manifest_sha256, args.output)), + 'Worker and parent arguments cannot be combined') + result = run_worker(args.worker_job, args.worker_job_sha256) + else: + require(args.fixtures and args.fixture_manifest_sha256 and args.output + and not args.worker_job_sha256, 'Require --fixtures --fixture-manifest-sha256 --output') + result = run(args.fixtures, args.fixture_manifest_sha256, args.output) + print(json.dumps(dict(complete=True, status=result.get('status', 'worker-returned'))), flush=True) + + +if __name__ == '__main__': + main() diff --git a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_brain_replay.py b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_brain_replay.py new file mode 100644 index 0000000..55ba2e4 --- /dev/null +++ b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_brain_replay.py @@ -0,0 +1,161 @@ +"""Host-only authority, exact-comparison, ordering and process receipt tests.""" +from __future__ import annotations + +import hashlib +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +import numpy as np + +import brain_replay as replay + + +class ReplayTests(unittest.TestCase): + def test_fixed_forward_reverse_order(self): + result = replay.fixture_order(14) + self.assertEqual(len(result), 28) + self.assertEqual(result[:14], [('forward', i) for i in range(14)]) + self.assertEqual(result[14:], [('reverse', i) for i in range(13, -1, -1)]) + for count in (0, 13, 15): + with self.assertRaises(ValueError): + replay.fixture_order(count) + + def test_exact_comparison_detects_signed_zero_dtype_shape_and_bit_changes(self): + expected = {'v': np.array([0., 1.], np.float64)} + self.assertTrue(replay.compare_arrays(expected, expected, ('v',))['v']['bitwise_exact']) + for array in (np.array([-0., 1.], np.float64), np.array([0., 1.], np.float32), + np.array([[0., 1.]], np.float64), + np.array([0., np.nextafter(1., 2.)], np.float64)): + self.assertFalse(replay.compare_arrays({'v': array}, expected, ('v',))['v']['bitwise_exact']) + with self.assertRaisesRegex(ValueError, 'schema'): + replay.compare_arrays({}, expected, ('v',)) + + def test_checkpoint_comparison_includes_all_arrays_requests_and_environment(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + actual_dir, old_dir = root / 'actual', root / 'old' + actual_dir.mkdir() + old_dir.mkdir() + for name in ('requests.csv', 'finite-audit.json', 'apl-pending-tail.float32.npy'): + (actual_dir / name).write_bytes(b'fixed exact reference') + (old_dir / name).write_bytes(b'fixed exact reference') + def record(): + return dict(state={name: np.zeros(2, np.float64 if name in ('v', 'g') else np.int32) + for name in replay.STATE_KEYS}, + terminal={'spike_ring': np.zeros((19, 2), np.uint8), + 'apl_state': np.zeros(2, np.float32), + 'apl_pending': np.zeros((18, 2), np.float32), + 'next_tick': np.asarray(1200, np.int64)}, + summary={**{key: 'same' for key in replay.SUMMARY_KEYS}, + 'apl_component': {'environment': {'torch': 'fixed-original'}}}, + fresh=True, global_tick_offset=0) + actual, original = record(), record() + self.assertTrue(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact']) + actual['terminal']['spike_ring'][0, 1] = 1 + self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact']) + actual = record() + actual['summary']['apl_component']['environment']['torch'] = 'different' + self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact']) + actual = record() + actual['summary']['effective_weights_raw_sha256'] = 'different' + self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact']) + actual = record() + (actual_dir / 'requests.csv').write_bytes(b'different acceptance') + self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact']) + + def test_json_and_pin_authority_rejects_ambiguity_and_mutation(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + value = root / 'value.json' + value.write_text('{"key":1,"key":2}', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'Duplicate'): + replay.read(value) + value.write_text('{"key":NaN}', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'Nonfinite'): + replay.read(value) + value.write_text('{"key":1}', encoding='utf-8') + pins = {} + replay.pin_add(pins, value, replay.sha(value)) + replay.verify(pins) + with self.assertRaisesRegex(ValueError, 'Conflicting'): + replay.pin_add(pins, value, 'f' * 64) + value.write_text('{"key":2}', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'changed'): + replay.verify(pins) + with self.assertRaisesRegex(ValueError, 'leaves'): + replay.contained(root, str(root.parent / 'outside')) + + def test_loader_executes_only_verified_bytes_and_cannot_rebind_module(self): + with tempfile.TemporaryDirectory() as directory: + source = Path(directory) / 'source.py' + source.write_text('value = 42\n', encoding='utf-8') + name = '_brain_replay_host_test_unique_module' + try: + with self.assertRaisesRegex(ValueError, 'Unverified'): + replay.load_source(name, source, '0' * 64) + self.assertNotIn(name, sys.modules) + module = replay.load_source(name, source, replay.sha(source)) + self.assertEqual(module.value, 42) + with self.assertRaisesRegex(ValueError, 'repeated'): + replay.load_source(name, source, replay.sha(source)) + finally: + sys.modules.pop(name, None) + + def test_parent_environment_is_original_only(self): + with mock.patch.object(replay.sys, 'dont_write_bytecode', False): + with self.assertRaisesRegex(ValueError, 'original registered'): + replay.check_environment() + + +class ProcessReceiptTests(unittest.TestCase): + def paths(self, root): + for name in ('jobs', 'processes'): + (root / name).mkdir() + report_path = root / 'report.json' + replay.save(report_path, {'exact': True}) + result_path = root / 'result.json' + replay.save(result_path, dict(complete=True, all_bitwise_exact=True, lane='fresh', count=1, + records=[dict(report=str(report_path), sha256=replay.sha(report_path))])) + return dict(lane='fresh', runs=[['fresh', 0]], result_file=str(result_path)), root / 'jobs/fresh.json', root / 'processes' + + def test_child_receipt_requires_actual_return_and_keeps_outer_group(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + job, job_path, process_dir = self.paths(root) + process = mock.Mock(pid=4321, returncode=0) + process.wait.return_value = 0 + with mock.patch('brain_replay.subprocess.Popen', return_value=process) as popen: + receipt, result = replay.launch(job, job_path, process_dir, replay.time.time() + 10) + self.assertTrue(receipt['exit_independently_observed']) + self.assertEqual(receipt['actual_returncode'], 0) + self.assertEqual(receipt['child_pid'], 4321) + self.assertNotIn('start_new_session', popen.call_args.kwargs) + self.assertEqual(popen.call_args.kwargs['env']['CUDA_VISIBLE_DEVICES'], replay.GPU) + self.assertEqual(popen.call_args.args[0][0], str(replay.ORIGINAL_PYTHON)) + self.assertEqual(replay.read(process_dir / 'fresh.json')['actual_returncode'], 0) + self.assertTrue(result['all_bitwise_exact']) + + def test_timeout_stops_child_retains_real_failure_receipt_and_never_retries(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + job, job_path, process_dir = self.paths(root) + process = mock.Mock(pid=4321, returncode=-15) + process.wait.side_effect = [subprocess.TimeoutExpired('mock child', 1), -15, -15] + process.poll.return_value = None + with mock.patch('brain_replay.subprocess.Popen', return_value=process) as popen: + with self.assertRaisesRegex(ValueError, 'child failed'): + replay.launch(job, job_path, process_dir, replay.time.time() + 10) + self.assertEqual(popen.call_count, 1) + process.terminate.assert_called_once() + receipt = replay.read(process_dir / 'fresh.json') + self.assertTrue(receipt['timed_out']) + self.assertTrue(receipt['child_terminated_on_failure']) + self.assertEqual(receipt['actual_returncode'], -15) + + +if __name__ == '__main__': + unittest.main()