Digital Drosophila Archive
567 lines
33 KiB
Python
567 lines
33 KiB
Python
"""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()
|