Digital Drosophila Archive
771 lines
44 KiB
Python
771 lines
44 KiB
Python
"""Read-only profiling of the fixed 14 Phase 4AR update fixtures.
|
|
|
|
The original Python -B environment and unchanged original U/AM/J sources are
|
|
mandatory. No brain is run, training is resumed, or computed weight installed.
|
|
Outputs are exclusive files under one fresh directory outside frozen inputs.
|
|
|
|
Unprofiled lanes measure fresh U processes, two separate resident CPU
|
|
processes (U producer and original U.verify_receipt verifier), and fresh full AQ
|
|
validators. A further lane calls unchanged full E in fresh and resident CPU
|
|
processes, retaining all its original checks. The resident verifier does NOT replace AQ checkpoint/graph/source
|
|
validation. Production checks must not be removed on the basis of this replay.
|
|
Separate cProfile runs are diagnostic and excluded from timing comparisons.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import cProfile
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import pstats
|
|
import queue
|
|
import statistics
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
|
|
ROOT = Path('/data/digital-drosophila')
|
|
MODEL = ROOT / 'models/male-cns-lif'
|
|
ENV = MODEL / '.venv'
|
|
EXP = ROOT / 'experiments/phase4ar-normal-conditioning'
|
|
REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740'
|
|
COMMIT = '80aeb2093db655329774eff7b32fb6e417b8b40b'
|
|
SOURCE_SHA = dict(
|
|
conditioning_update_source='96efff678cff910654eb08e86f1a17789e4e5589e340345f419c7027a7d547e2',
|
|
eligibility_bridge_source='174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61',
|
|
phase4j_source='abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e')
|
|
E_PATH = MODEL / 'olfactory-conditioning-lab-v1/conditioning_epoch.py'
|
|
E_SHA = '757935a4815d9f625640df685184ed3fe44a8845323562796da7aa43cbb194e1'
|
|
INPUTS = ('step', 'neuron_idx', 'kc', 'candidate_edge_id', 'candidate_sources',
|
|
'candidate_targets', 'candidate_compartments', 'delta_before', 'base_candidate')
|
|
OUTPUTS = ('delta_after', 'sampled_trace', 'final_trace')
|
|
REPEATS = 3
|
|
SCHEMA = 'conditioning-pipeline-original-profile-v1'
|
|
CHILDREN = []
|
|
|
|
|
|
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 payload_sha(value):
|
|
return hashlib.sha256(canonical(value).encode()).hexdigest()
|
|
|
|
|
|
def unlinked(value):
|
|
path = Path(value)
|
|
require(path.is_absolute() and '..' not in path.parts, 'Explicit absolute 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 digest(path):
|
|
result = hashlib.sha256()
|
|
with unlinked(path).open('rb') as stream:
|
|
for data in iter(lambda: stream.read(8 << 20), b''):
|
|
result.update(data)
|
|
return result.hexdigest()
|
|
|
|
|
|
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(size == path.stat().st_size, 'File size changed during read')
|
|
return result
|
|
|
|
|
|
def check(path, pin):
|
|
observed = identity(path)
|
|
sha = pin if isinstance(pin, str) else pin['sha256']
|
|
require(observed['sha256'] == sha, 'File SHA differs: ' + str(path))
|
|
if isinstance(pin, dict) and 'bytes' in pin:
|
|
require(observed['bytes'] == pin['bytes'], 'File size differs: ' + str(path))
|
|
return observed
|
|
|
|
|
|
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 contained(root, name):
|
|
require(type(name) is str and '\\' not in name, 'Invalid relative fixture path')
|
|
relative = PurePosixPath(name)
|
|
require(not relative.is_absolute() and str(relative) == name and '..' not in relative.parts
|
|
and all(':' not in part for part in relative.parts), 'Escaping fixture path')
|
|
result = unlinked(root / name)
|
|
require(result.is_relative_to(root), 'Fixture path escaped root')
|
|
return result
|
|
|
|
|
|
def 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 same(left, right):
|
|
return left.dtype == right.dtype and left.shape == right.shape and left.tobytes() == right.tobytes()
|
|
|
|
|
|
def cpu_start():
|
|
require(sys.dont_write_bytecode and 'torch' not in sys.modules and 'runtime_light' not in sys.modules,
|
|
'Fresh Python -B CPU process required')
|
|
require(Path(sys.prefix) == ENV, 'Use the unchanged original registered .venv')
|
|
os.environ['CUDA_VISIBLE_DEVICES'] = ''
|
|
os.environ['PYTHONDONTWRITEBYTECODE'] = '1'
|
|
|
|
|
|
def load_source(name, path, sha):
|
|
raw = unlinked(path).read_bytes()
|
|
require(hashlib.sha256(raw).hexdigest() == sha and name not in sys.modules, 'Source/module identity differs')
|
|
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 read_manifest(fixtures, expected):
|
|
path = fixtures / 'manifest.json'
|
|
check(path, expected)
|
|
manifest = read(path)
|
|
require(manifest['schema'] == 'conditioning-acceleration-fixtures-v1' and manifest['complete'] is True
|
|
and manifest['registration_sha256'] == REG_SHA and manifest['source_code_commit'] == COMMIT
|
|
and manifest['fixture_count'] == len(manifest['fixtures']) == 14,
|
|
'Exact sealed fourteen-fixture authority required')
|
|
require(len({f['fixture_id'] for f in manifest['fixtures']}) == 14, 'Duplicate fixture identity')
|
|
require(manifest['selection_sha256'] == manifest['files']['selection.json']['sha256'], 'Selection authority differs')
|
|
return manifest
|
|
|
|
|
|
def verify_suite(fixtures, expected):
|
|
manifest = read_manifest(fixtures, expected)
|
|
for relative, pin in manifest['files'].items():
|
|
check(contained(fixtures, relative), pin)
|
|
registration = EXP / 'protocol/preregistered.json'
|
|
check(registration, REG_SHA)
|
|
reg = read(registration)
|
|
require(reg['source_code_commit'] == COMMIT, 'Registered scientific source commit differs')
|
|
original_pins = {str(registration): REG_SHA, **reg['source_sha256']}
|
|
for entry in manifest['fixtures']:
|
|
directory = contained(fixtures, entry['fixture_id'])
|
|
for path, pin in read(directory / 'source-file-pins.json').items():
|
|
prior = original_pins.get(path)
|
|
require(prior is None or (prior if isinstance(prior, str) else prior['sha256']) == pin['sha256'],
|
|
'Conflicting immutable input pin')
|
|
original_pins[path] = pin
|
|
require(entry['exact_stored_delta_and_trace_verified'] is True
|
|
and entry['full_canonical_KC_count'] == 4064 and entry['full_candidate_edge_count'] == 14232
|
|
and entry['original_validator_verified_full_graph_mapping'] is True, 'Unvalidated fixture identity')
|
|
for path, pin in original_pins.items():
|
|
check(path, pin)
|
|
validator = MODEL / 'conditioning-execution-v1/validate_update_cpu.py'
|
|
original_pins[str(validator)] = check(validator, reg['input_sha256'][str(validator)])
|
|
require(reg['input_sha256'][str(E_PATH)] == E_SHA, 'Registered E source pin differs')
|
|
original_pins[str(E_PATH)] = check(E_PATH, E_SHA)
|
|
return manifest, original_pins
|
|
|
|
|
|
def fixture_inputs(fixtures, manifest, entry):
|
|
"""Each child independently reads/checks/owns its input and reference arrays."""
|
|
import numpy as np
|
|
directory = contained(fixtures, entry['fixture_id'])
|
|
required = (entry['npz'], entry['command'], entry['fixture_id'] + '/reference-sources.json',
|
|
entry['fixture_id'] + '/stored-update-receipt.json')
|
|
for relative in required:
|
|
check(contained(fixtures, relative), manifest['files'][relative])
|
|
require(manifest['files'][entry['npz']]['sha256'] == entry['npz_sha256']
|
|
and manifest['files'][entry['command']]['sha256'] == entry['command_sha256'], 'Fixture entry/file pins differ')
|
|
with np.load(contained(fixtures, entry['npz']), allow_pickle=False) as archive:
|
|
arrays = {name: archive[name].copy() for name in archive.files}
|
|
inputs = {name: arrays[name] for name in INPUTS}
|
|
require({name: evidence(array) for name, array in inputs.items()} == entry['input_arrays'],
|
|
'Loaded fixture full input bytes differ')
|
|
for name, pin in entry['output_arrays'].items():
|
|
require(evidence(arrays[name]) == pin, 'Loaded fixture reference bytes differ: ' + name)
|
|
require(len(inputs['kc']) == 4064 and len(inputs['candidate_edge_id']) == 14232,
|
|
'Incomplete original canonical array identity')
|
|
command = read(contained(fixtures, entry['command']))
|
|
require(command['trace_backend'] == 'torch', 'Original Torch CPU backend required')
|
|
sources = read(directory / 'reference-sources.json')
|
|
require(set(sources) == set(SOURCE_SHA), 'Unexpected original source roles')
|
|
for role, pin in sources.items():
|
|
require(pin['sha256'] == SOURCE_SHA[role], 'Frozen source SHA differs')
|
|
check(pin['path'], pin['sha256'])
|
|
kwargs = dict(inputs, **command, bridge_source=sources['eligibility_bridge_source']['path'],
|
|
phase4j_source=sources['phase4j_source']['path'])
|
|
receipt = read(directory / 'stored-update-receipt.json')
|
|
require(receipt['payload_sha256'] == entry['original_update_payload_sha256'], 'Stored receipt identity differs')
|
|
for name in OUTPUTS:
|
|
require(evidence(arrays['reference_' + name]) == receipt['outputs'][name], 'Receipt/reference byte evidence differs')
|
|
require(same(arrays['reference_delta_after'], arrays['stored_delta_after']), 'Stored delta differs from CPU reference')
|
|
for array in arrays.values():
|
|
array.setflags(write=False)
|
|
return dict(inputs=inputs, arrays=arrays, kwargs=kwargs, sources=sources, receipt=receipt, command=command)
|
|
|
|
|
|
def original_locations(fixtures, entry):
|
|
directory = contained(fixtures, entry['fixture_id'])
|
|
authority = read(directory / 'authority.json')
|
|
pins = read(directory / 'source-file-pins.json')
|
|
authority_paths = [path for path, pin in pins.items()
|
|
if path.endswith('.authority.json') and pin['sha256'] == entry['authority_sha256']]
|
|
require(len(authority_paths) == 1, 'Original authority location is ambiguous')
|
|
authority_path = authority_paths[0]
|
|
check(authority_path, entry['authority_sha256'])
|
|
require(read(authority_path) == authority, 'Copied/original authority differs')
|
|
task_key = Path(authority_path).name[:-len('.authority.json')]
|
|
update_dir = EXP / 'updates' / task_key
|
|
chain_path = update_dir / 'chain-receipt.json'
|
|
chain_sha = pins[str(chain_path)]['sha256']
|
|
return authority_path, authority, update_dir, chain_sha
|
|
|
|
|
|
def assert_epoch_output(directory, chain, fixture, entry):
|
|
import numpy as np
|
|
require(chain['complete'] is True and chain['coordinator_source_sha256'] == E_SHA
|
|
and chain['authority_sha256'] == entry['authority_sha256']
|
|
and chain['weights_installed'] is False and chain['brain_executed'] is False
|
|
and chain['input_and_checkpoint_bytes_reverified'] is True
|
|
and chain['candidate_endpoints_equal_graph'] is True
|
|
and chain['KC_equals_supplied_canonical_membership'] is True,
|
|
'Original full E validation chain incomplete')
|
|
receipt = read(directory / 'update-receipt.json')
|
|
require(canonical(receipt) == canonical(fixture['receipt']), 'Full original E receipt differs from sealed CPU reference')
|
|
with np.load(directory / 'delta-after.npz', allow_pickle=False) as archive:
|
|
require(set(archive.files) == {'edge_id', 'plastic_delta_mv'}, 'Full E overlay NPZ schema differs')
|
|
require(same(archive['edge_id'], fixture['inputs']['candidate_edge_id'])
|
|
and same(archive['plastic_delta_mv'], fixture['arrays']['reference_delta_after']),
|
|
'Full original E candidate/delta bytes differ')
|
|
for name in OUTPUTS:
|
|
require(receipt['outputs'][name] == evidence(fixture['arrays']['reference_' + name]),
|
|
'Full E trace/delta receipt does not bind full reference bytes')
|
|
for filename, sha in chain['output_files_sha256'].items():
|
|
check(directory / filename, sha)
|
|
return receipt
|
|
|
|
|
|
def assert_update(result, fixture):
|
|
for name in OUTPUTS:
|
|
require(same(getattr(result, name), fixture['arrays']['reference_' + name]), 'Full array mismatch: ' + name)
|
|
require(canonical(result.receipt) == canonical(fixture['receipt']), 'Original complete receipt mismatch')
|
|
require({name: evidence(value) for name, value in fixture['inputs'].items()} == result.receipt['input_arrays'],
|
|
'Resident input bytes changed')
|
|
snapshots = fixture['arrays']['reference_snapshots']
|
|
if fixture['command']['us_tick'] is None:
|
|
require(snapshots.shape == (0, 4064) and not result.sampled_trace.view('uint32').any(),
|
|
'No-US snapshot/+0 rule differs')
|
|
else:
|
|
require(snapshots.shape == (1, 4064) and same(snapshots[0], result.sampled_trace),
|
|
'After-current-tick sampled snapshot differs')
|
|
|
|
|
|
def pack_array(array):
|
|
return dict(dtype=array.dtype.str, shape=list(array.shape),
|
|
data_base64=base64.b64encode(array.tobytes(order='C')).decode('ascii'))
|
|
|
|
|
|
def compare_packet(packet, reference):
|
|
require(packet['dtype'] == reference.dtype.str and packet['shape'] == list(reference.shape),
|
|
'Producer packet array identity differs')
|
|
raw = base64.b64decode(packet['data_base64'], validate=True)
|
|
require(raw == reference.tobytes(order='C'), 'Producer full array bytes differ from independent reference')
|
|
|
|
|
|
def verify_response(value, previous, sequence, role):
|
|
require(value['role'] == role and value['sequence'] == sequence
|
|
and value['previous_response_sha256'] == previous and value['complete'] is True,
|
|
'Worker response chain/role differs')
|
|
expected = payload_sha({key: item for key, item in value.items() if key != 'response_sha256'})
|
|
require(expected == value['response_sha256'], 'Worker response payload hash differs')
|
|
return expected
|
|
|
|
|
|
def profile_category(filename, name):
|
|
"""Exclusive SELF-time categories; cumulative function rows stay separate."""
|
|
if 'importlib' in filename or name in ('_imp.create_dynamic', '_imp.exec_dynamic'):
|
|
return 'module_import'
|
|
if '::trace-expression-only' in filename or any("'" + op + "'" in name for op in ('mul_', 'add_', 'clamp_max_', 'zero_')):
|
|
return 'trace_expression_and_tensor_ops'
|
|
if filename.endswith('/ast.py') or filename.endswith('\\ast.py') or name in ('_load_original', '_load_bridge') or 'builtins.compile' in name:
|
|
return 'source_parse_compile_load'
|
|
if name in ('trace_from_events', 'apply_eligibility', 'apply_v2'):
|
|
return 'trace_apply_python_self'
|
|
if any(token in name for token in ('read_bytes', 'readinto', 'readall', 'HASH', 'openssl_sha256', '_pinned_bytes', '_evidence')):
|
|
return 'file_reads_and_hashing'
|
|
return 'other_validation_arrays_and_runtime'
|
|
|
|
|
|
def profile_rows(profiler):
|
|
rows, categories = [], {}
|
|
for (filename, line, name), (primitive, calls, own, cumulative, _) in pstats.Stats(profiler).stats.items():
|
|
category = profile_category(filename, name)
|
|
categories[category] = categories.get(category, 0.) + own
|
|
rows.append(dict(file=filename, line=line, function=name, primitive_calls=primitive, calls=calls,
|
|
self_seconds=own, cumulative_seconds=cumulative, category=category))
|
|
return dict(self_time_categories_seconds=categories,
|
|
functions=sorted(rows, key=lambda row: row['cumulative_seconds'], reverse=True),
|
|
interpretation='diagnostic Python cProfile overhead included; self categories are exclusive, cumulative rows overlap; not standalone pure-math latency')
|
|
|
|
|
|
def worker(fixtures, expected, role, worker_output_root=None):
|
|
cpu_start()
|
|
started = time.perf_counter()
|
|
manifest = read_manifest(fixtures, expected)
|
|
entries = {entry['fixture_id']: entry for entry in manifest['fixtures']}
|
|
t = time.perf_counter()
|
|
import numpy as np
|
|
numpy_seconds = time.perf_counter() - t
|
|
first = contained(fixtures, manifest['fixtures'][0]['fixture_id'])
|
|
check(first / 'reference-sources.json', manifest['files'][manifest['fixtures'][0]['fixture_id'] + '/reference-sources.json'])
|
|
sources = read(first / 'reference-sources.json')
|
|
source = sources['conditioning_update_source']
|
|
require(source['sha256'] == SOURCE_SHA['conditioning_update_source'], 'Original U source differs')
|
|
t = time.perf_counter()
|
|
update = load_source('_pipeline_original_U_' + role, source['path'], source['sha256'])
|
|
source_seconds = time.perf_counter() - t
|
|
epoch = None
|
|
epoch_import_seconds = 0.
|
|
if role == 'epoch':
|
|
require(worker_output_root is not None and unlinked(worker_output_root).is_dir(), 'Explicit existing full E output root required')
|
|
t = time.perf_counter()
|
|
epoch = load_source('_pipeline_original_E', E_PATH, E_SHA)
|
|
epoch_import_seconds = time.perf_counter() - t
|
|
print(canonical(dict(event='ready', role=role, pid=os.getpid(), fixture_manifest_sha256=expected,
|
|
setup_seconds=time.perf_counter() - started, numpy_import_seconds=numpy_seconds,
|
|
original_U_source_import_seconds=source_seconds, torch_loaded='torch' in sys.modules,
|
|
original_E_source_import_seconds=epoch_import_seconds,
|
|
CUDA_VISIBLE_DEVICES='', source_pins=sources)), flush=True)
|
|
cache, previous, sequence = {}, None, 0
|
|
producer_previous, producer_sequence, producer_pid = None, 0, None
|
|
for line in sys.stdin:
|
|
request = json.loads(line)
|
|
if request.get('action') == 'stop':
|
|
break
|
|
require(request.get('action') in ('run', 'profile'), 'Unknown request action')
|
|
require(role != 'verifier' or request['action'] == 'run', 'Verifier profiling is not an authorized lane')
|
|
request_started = time.perf_counter()
|
|
entry = entries[request['fixture_id']]
|
|
t = time.perf_counter()
|
|
cached = request['fixture_id'] in cache
|
|
if not cached:
|
|
cache[request['fixture_id']] = fixture_inputs(fixtures, manifest, entry)
|
|
fixture = cache[request['fixture_id']]
|
|
load_seconds = time.perf_counter() - t
|
|
require(fixture['sources'] == sources, 'Fixture source closure differs from resident original module')
|
|
verify_seconds = 0.
|
|
packets = None
|
|
receipt = fixture['receipt']
|
|
diagnostic = None
|
|
epoch_output = None
|
|
if role == 'epoch':
|
|
require(request['action'] == 'run', 'Full E profiling uses unprofiled original API only')
|
|
epoch_output = unlinked(request['epoch_output'])
|
|
require(epoch_output.parent == worker_output_root and not epoch_output.exists(), 'Full E output must be a fresh direct child')
|
|
authority_path, _, _, _ = original_locations(fixtures, entry)
|
|
t = time.perf_counter()
|
|
chain = epoch.postprocess_epoch(authority_path, entry['authority_sha256'], epoch_output)
|
|
api_seconds = time.perf_counter() - t
|
|
t = time.perf_counter()
|
|
receipt = assert_epoch_output(epoch_output, chain, fixture, entry)
|
|
verify_seconds = time.perf_counter() - t
|
|
elif role == 'verifier':
|
|
produced = request['producer_response']
|
|
producer_sequence += 1
|
|
producer_previous = verify_response(produced, producer_previous, producer_sequence, 'producer')
|
|
require(produced['fixture_id'] == request['fixture_id'] and produced['request_id'] == request['request_id'],
|
|
'Producer/verifier task pairing differs')
|
|
if producer_pid is None:
|
|
producer_pid = produced['pid']
|
|
require(produced['pid'] == producer_pid and producer_pid != os.getpid(), 'Verifier must be a distinct stable OS process')
|
|
t = time.perf_counter()
|
|
require(canonical(produced['receipt']) == canonical(receipt), 'Producer receipt differs from sealed CPU reference')
|
|
for name in OUTPUTS:
|
|
compare_packet(produced['arrays'][name], fixture['arrays']['reference_' + name])
|
|
packet_seconds = time.perf_counter() - t
|
|
t = time.perf_counter()
|
|
require(update.verify_receipt(produced['receipt'], **fixture['kwargs']) is True, 'Original U receipt verification failed')
|
|
api_seconds = time.perf_counter() - t
|
|
verify_seconds = packet_seconds
|
|
else:
|
|
profiler = cProfile.Profile() if request['action'] == 'profile' else None
|
|
t = time.perf_counter()
|
|
if profiler:
|
|
profiler.enable()
|
|
try:
|
|
result = update.update_epoch(**fixture['kwargs'])
|
|
finally:
|
|
if profiler:
|
|
profiler.disable()
|
|
api_seconds = time.perf_counter() - t
|
|
t = time.perf_counter()
|
|
assert_update(result, fixture)
|
|
packets = {name: pack_array(getattr(result, name)) for name in OUTPUTS}
|
|
verify_seconds = time.perf_counter() - t
|
|
receipt = result.receipt
|
|
if profiler:
|
|
diagnostic = profile_rows(profiler)
|
|
import torch
|
|
require(not torch.cuda.is_initialized() and os.environ['CUDA_VISIBLE_DEVICES'] == '', 'CPU worker initialized CUDA')
|
|
require({name: evidence(array) for name, array in fixture['inputs'].items()} == entry['input_arrays'],
|
|
'Resident inputs differ after call')
|
|
sequence += 1
|
|
response = dict(schema=SCHEMA + '-response', complete=True, role=role, pid=os.getpid(), sequence=sequence,
|
|
previous_response_sha256=previous, request_id=request['request_id'], fixture_id=request['fixture_id'],
|
|
cached_inputs=cached, input_load_verify_seconds=load_seconds, original_api_seconds=api_seconds,
|
|
reference_checks_and_packet_seconds=verify_seconds, worker_request_seconds=time.perf_counter() - request_started,
|
|
full_array_reference_exact=True, original_receipt_exact=True, input_bytes_unchanged=True,
|
|
cuda_initialized=False, receipt=receipt, arrays=packets, cprofile=diagnostic,
|
|
epoch_output=None if epoch_output is None else str(epoch_output),
|
|
independently_verified_producer_response_sha256=producer_previous if role == 'verifier' else None,
|
|
versions=dict(python=sys.version, numpy=np.__version__, torch=torch.__version__),
|
|
verification_scope=('unchanged full E.postprocess_epoch including all original source/graph/checkpoint checks; exact delta bytes and full receipt trace hashes'
|
|
if role == 'epoch' else 'independent original U.verify_receipt recomputation plus exact full producer-array/reference comparison; NOT full AQ graph/checkpoint validation'
|
|
if role == 'verifier' else 'unchanged original U.update_epoch; all output array bytes equal sealed CPU reference'))
|
|
response['response_sha256'] = payload_sha(response)
|
|
previous = response['response_sha256']
|
|
print(canonical(response), flush=True)
|
|
for fixture in cache.values():
|
|
for source in fixture['sources'].values():
|
|
check(source['path'], source['sha256'])
|
|
check(fixtures / 'manifest.json', expected)
|
|
|
|
|
|
class Child:
|
|
def __init__(self, output, name, fixtures, expected, role, worker_output_root=None):
|
|
self.name, self.output, self.role = name, output, role
|
|
self.previous, self.sequence, self.closed = None, 0, False
|
|
self.started = time.perf_counter()
|
|
self.stderr = (output / (name + '.stderr.log')).open('xb')
|
|
self.responses = (output / (name + '.responses.jsonl')).open('x', encoding='utf-8', newline='\n')
|
|
self.argv = [sys.executable, '-B', str(Path(__file__).absolute()), '--fixtures', str(fixtures),
|
|
'--fixture-manifest-sha256', expected, '--worker', role]
|
|
if worker_output_root is not None:
|
|
self.argv += ['--worker-output-root', str(worker_output_root)]
|
|
self.process = subprocess.Popen(self.argv, cwd=output, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=self.stderr, text=True, encoding='utf-8', bufsize=1,
|
|
env=dict(os.environ, CUDA_VISIBLE_DEVICES='', PYTHONDONTWRITEBYTECODE='1'))
|
|
CHILDREN.append(self)
|
|
self.lines = queue.Queue()
|
|
def pump():
|
|
try:
|
|
for line in self.process.stdout:
|
|
self.lines.put(line)
|
|
finally:
|
|
self.lines.put(None)
|
|
self.reader = threading.Thread(target=pump, daemon=True)
|
|
self.reader.start()
|
|
self.ready = self._read()
|
|
require(self.ready['event'] == 'ready' and self.ready['role'] == role
|
|
and self.ready['pid'] == self.process.pid and self.ready['fixture_manifest_sha256'] == expected
|
|
and self.ready['torch_loaded'] is False, 'Fresh CPU worker startup differs')
|
|
self.startup_wall = time.perf_counter() - self.started
|
|
|
|
def _read(self):
|
|
try:
|
|
line = self.lines.get(timeout=120)
|
|
except queue.Empty as error:
|
|
raise TimeoutError(self.name + ' response exceeded 120 seconds') from error
|
|
require(bool(line), self.name + ' exited without a complete response; inspect stderr')
|
|
self.responses.write(line)
|
|
self.responses.flush()
|
|
return json.loads(line)
|
|
|
|
def request(self, value):
|
|
started = time.perf_counter()
|
|
self.process.stdin.write(canonical(value) + '\n')
|
|
self.process.stdin.flush()
|
|
response = self._read()
|
|
wall = time.perf_counter() - started
|
|
self.sequence += 1
|
|
self.previous = verify_response(response, self.previous, self.sequence, self.role)
|
|
require(response['pid'] == self.process.pid and response['fixture_id'] == value['fixture_id']
|
|
and response['request_id'] == value['request_id'], 'Worker response pairing differs')
|
|
return response, wall
|
|
|
|
def close(self):
|
|
if self.closed:
|
|
return self.receipt
|
|
self.process.stdin.write(canonical(dict(action='stop')) + '\n')
|
|
self.process.stdin.flush()
|
|
self.process.stdin.close()
|
|
code = self.process.wait(timeout=30)
|
|
self.reader.join(timeout=2)
|
|
self.process.stdout.close()
|
|
self.stderr.close()
|
|
self.responses.close()
|
|
self.receipt = dict(argv=self.argv, pid=self.process.pid, exit_code=code,
|
|
actual_exit_observed=True, startup_wall_seconds=self.startup_wall,
|
|
process_wall_seconds=time.perf_counter() - self.started, ready=self.ready,
|
|
responses=self.sequence, final_response_sha256=self.previous,
|
|
stdout_sha256=digest(self.output / (self.name + '.responses.jsonl')),
|
|
stderr_sha256=digest(self.output / (self.name + '.stderr.log')))
|
|
save(self.output / (self.name + '-process.json'), self.receipt)
|
|
self.closed = True
|
|
require(code == 0, self.name + ' failed; output retained')
|
|
return self.receipt
|
|
|
|
def abort(self):
|
|
"""Only this exact spawned child is stopped; partial evidence is retained."""
|
|
if self.closed:
|
|
return
|
|
terminate_sent = False
|
|
kill_sent = False
|
|
if self.process.poll() is None:
|
|
self.process.terminate()
|
|
terminate_sent = True
|
|
try:
|
|
code = self.process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
self.process.kill()
|
|
kill_sent = True
|
|
code = self.process.wait(timeout=30)
|
|
self.reader.join(timeout=2)
|
|
for stream in (self.process.stdin, self.process.stdout, self.stderr, self.responses):
|
|
if stream is not None and not stream.closed:
|
|
stream.close()
|
|
save(self.output / (self.name + '-abort-process.json'), dict(argv=self.argv, pid=self.process.pid,
|
|
actual_exit_observed=True, exit_code=code, terminate_sent=terminate_sent, kill_sent=kill_sent,
|
|
partial_evidence_retained=True, automatic_retry=False))
|
|
self.closed = True
|
|
|
|
|
|
def cleanup_children():
|
|
errors = []
|
|
for child in reversed(CHILDREN):
|
|
try:
|
|
child.abort()
|
|
except Exception as error:
|
|
errors.append(type(error).__name__ + ': ' + str(error))
|
|
require(not errors, 'Child cleanup errors: ' + '; '.join(errors))
|
|
|
|
|
|
def compact_response(response, wall):
|
|
return {**{key: value for key, value in response.items() if key not in ('arrays', 'receipt', 'cprofile')},
|
|
'original_receipt_payload_sha256': response['receipt']['payload_sha256'],
|
|
'parent_roundtrip_seconds': wall}
|
|
|
|
|
|
def distribution(values):
|
|
require(bool(values), 'Empty timing distribution')
|
|
return dict(count=len(values), minimum=min(values), median=statistics.median(values),
|
|
maximum=max(values), sum=sum(values), mean=statistics.mean(values))
|
|
|
|
|
|
def full_validation(fixtures, entry, output, reg, produced_update_dir):
|
|
directory = contained(fixtures, entry['fixture_id'])
|
|
authority_path, _, historical_update_dir, _ = original_locations(fixtures, entry)
|
|
update_dir = unlinked(produced_update_dir)
|
|
require(update_dir.is_dir() and update_dir != historical_update_dir
|
|
and not update_dir.is_relative_to(EXP), 'Full AQ must validate newly produced E artifacts outside AR')
|
|
filenames = ('chain-receipt.json', 'update-receipt.json', 'delta-after.npz', 'overlay-manifest.json')
|
|
require(set(path.name for path in update_dir.iterdir()) == set(filenames), 'New E output inventory differs')
|
|
update_hashes = {name: digest(update_dir / name) for name in filenames}
|
|
chain_sha = update_hashes['chain-receipt.json']
|
|
validator = MODEL / 'conditioning-execution-v1/validate_update_cpu.py'
|
|
check(validator, reg['input_sha256'][str(validator)])
|
|
receipt_path = output / (entry['fixture_id'] + '.validation.json')
|
|
argv = [sys.executable, '-B', str(validator), '--authority', authority_path,
|
|
'--authority-sha256', entry['authority_sha256'], '--update-dir', str(update_dir),
|
|
'--chain-sha256', chain_sha, '--receipt', str(receipt_path)]
|
|
started = time.perf_counter()
|
|
with (output / (entry['fixture_id'] + '.stdout.log')).open('xb') as stdout, \
|
|
(output / (entry['fixture_id'] + '.stderr.log')).open('xb') as stderr:
|
|
process = subprocess.run(argv, cwd=output, stdout=stdout, stderr=stderr,
|
|
env=dict(os.environ, CUDA_VISIBLE_DEVICES='', PYTHONDONTWRITEBYTECODE='1'), timeout=120, check=False)
|
|
wall = time.perf_counter() - started
|
|
record = dict(fixture_id=entry['fixture_id'], argv=argv, exit_code=process.returncode,
|
|
actual_exit_observed=True, subprocess_wall_seconds=wall,
|
|
validated_update_dir=str(update_dir), validated_new_E_artifact=True,
|
|
update_files_sha256=update_hashes)
|
|
save(output / (entry['fixture_id'] + '.process.json'), record)
|
|
require(process.returncode == 0, 'Original AQ validation failed; outputs retained')
|
|
expected = read(directory / 'fresh-cpu-validation.json')
|
|
# E's new NPZ/container hashes may differ while its scientific array bytes
|
|
# match. Only these independently observed output-file bindings may vary.
|
|
expected.update(update_files_sha256=update_hashes,
|
|
after_overlay_sha256=update_hashes['delta-after.npz'],
|
|
after_manifest_sha256=update_hashes['overlay-manifest.json'])
|
|
require(read(receipt_path) == expected,
|
|
'Fresh full AQ receipt differs from original scientific/source fields and new E file bindings')
|
|
require(update_hashes == {name: digest(update_dir / name) for name in filenames}, 'New E artifacts changed during full AQ validation')
|
|
record['full_original_AQ_receipt_exact_to_expected_binding'] = True
|
|
record['receipt_sha256'] = digest(receipt_path)
|
|
return record
|
|
|
|
|
|
def profile(fixtures, expected, output, repeats):
|
|
cpu_start()
|
|
require(repeats == REPEATS, 'This fixed engineering experiment requires exactly three resident passes')
|
|
started = time.perf_counter()
|
|
fixtures, output = unlinked(fixtures), unlinked(output)
|
|
own = identity(Path(__file__).absolute())
|
|
manifest, original_pins = verify_suite(fixtures, expected)
|
|
protected = {fixtures, EXP, MODEL, *[Path(path).parent for path in original_pins]}
|
|
require(not output.exists() and output.parent.is_dir()
|
|
and not any(output.is_relative_to(parent) for parent in protected),
|
|
'Output must be a fresh independent directory with an existing parent')
|
|
output.mkdir()
|
|
save(output / 'experiment.json', dict(schema=SCHEMA, fixture_manifest_sha256=expected,
|
|
fixture_order=[entry['fixture_id'] for entry in manifest['fixtures']], repeats=repeats,
|
|
selection='all previously sealed 14 fixtures, fixed existing manifest order, no outcome selection',
|
|
original_source_commit=COMMIT, registration_sha256=REG_SHA,
|
|
source_pins=original_pins, implementation_source=dict(path=str(Path(__file__).absolute()), **own),
|
|
scope='read-only fixed-input engineering replay; no installation, brain, training or learning claim',
|
|
resident_verifier_scope='original U.verify_receipt only; cannot substitute for original AQ full graph/checkpoint validator'))
|
|
cold_dir, resident_dir, aq_dir, diagnostic_dir, epoch_dir = [output / name for name in ('cold', 'resident', 'full-aq', 'diagnostic', 'full-E')]
|
|
for directory in (cold_dir, resident_dir, aq_dir, diagnostic_dir, epoch_dir):
|
|
directory.mkdir()
|
|
cold = []
|
|
for entry in manifest['fixtures']:
|
|
child = Child(cold_dir, entry['fixture_id'], fixtures, expected, 'producer')
|
|
response, wall = child.request(dict(action='run', request_id='cold/' + entry['fixture_id'], fixture_id=entry['fixture_id']))
|
|
exit_receipt = child.close()
|
|
cold.append(dict(compact_response(response, wall), process=exit_receipt))
|
|
print(canonical(dict(lane='cold-original-U', fixture_id=entry['fixture_id'], complete=True)), flush=True)
|
|
producer = Child(resident_dir, 'producer', fixtures, expected, 'producer')
|
|
verifier = Child(resident_dir, 'verifier', fixtures, expected, 'verifier')
|
|
require(producer.process.pid != verifier.process.pid, 'Independent verifier process required')
|
|
resident = []
|
|
for round_index in range(repeats):
|
|
for entry in manifest['fixtures']:
|
|
request = dict(action='run', request_id=f'round-{round_index:02d}/' + entry['fixture_id'], fixture_id=entry['fixture_id'])
|
|
pair_started = time.perf_counter()
|
|
produced, producer_wall = producer.request(request)
|
|
verified, verifier_wall = verifier.request(dict(request, producer_response=produced))
|
|
require(verified['independently_verified_producer_response_sha256'] == produced['response_sha256'],
|
|
'Independent verifier receipt chain mismatch')
|
|
resident.append(dict(round_index=round_index, fixture_id=entry['fixture_id'],
|
|
producer=compact_response(produced, producer_wall), verifier=compact_response(verified, verifier_wall),
|
|
pair_parent_wall_seconds=time.perf_counter() - pair_started))
|
|
print(canonical(dict(lane='resident-separate-original-U-and-verifier', round_index=round_index, complete=True)), flush=True)
|
|
producer_exit, verifier_exit = producer.close(), verifier.close()
|
|
epoch_artifacts = epoch_dir / 'new-updates'
|
|
epoch_artifacts.mkdir()
|
|
first = manifest['fixtures'][0]
|
|
epoch_cold_worker = Child(epoch_dir, 'cold-first', fixtures, expected, 'epoch', epoch_artifacts)
|
|
epoch_cold_response, epoch_cold_wall = epoch_cold_worker.request(dict(action='run', request_id='full-E-cold/' + first['fixture_id'],
|
|
fixture_id=first['fixture_id'], epoch_output=str(epoch_artifacts / ('cold-' + first['fixture_id']))))
|
|
epoch_cold_exit = epoch_cold_worker.close()
|
|
epoch_cold = dict(compact_response(epoch_cold_response, epoch_cold_wall), process=epoch_cold_exit)
|
|
print(canonical(dict(lane='fresh-full-original-E', fixture_id=first['fixture_id'], complete=True)), flush=True)
|
|
epoch_worker = Child(epoch_dir, 'resident', fixtures, expected, 'epoch', epoch_artifacts)
|
|
epoch_resident = []
|
|
for round_index in range(2):
|
|
for entry in manifest['fixtures']:
|
|
response, wall = epoch_worker.request(dict(action='run', request_id=f'full-E-{round_index:02d}/' + entry['fixture_id'],
|
|
fixture_id=entry['fixture_id'], epoch_output=str(epoch_artifacts / f'round-{round_index:02d}-{entry["fixture_id"]}')))
|
|
epoch_resident.append(dict(round_index=round_index, **compact_response(response, wall)))
|
|
print(canonical(dict(lane='resident-full-original-E', round_index=round_index,
|
|
fixture_id=entry['fixture_id'], complete=True)), flush=True)
|
|
epoch_exit = epoch_worker.close()
|
|
full = []
|
|
reg = read(EXP / 'protocol/preregistered.json')
|
|
for entry in manifest['fixtures']:
|
|
full.append(full_validation(fixtures, entry, aq_dir, reg,
|
|
epoch_artifacts / f'round-01-{entry["fixture_id"]}'))
|
|
print(canonical(dict(lane='full-original-AQ', fixture_id=entry['fixture_id'], complete=True)), flush=True)
|
|
# Independent diagnostics: first call includes cold Torch import; another
|
|
# full fixed pass has warm imports. These calls never enter speed medians.
|
|
diagnostic = Child(diagnostic_dir, 'profiler', fixtures, expected, 'producer')
|
|
diagnostics = []
|
|
first = manifest['fixtures'][0]
|
|
schedule = [('cold-fixed-first', first), *[('warm-fixed-all', entry) for entry in manifest['fixtures']]]
|
|
for index, (temperature, entry) in enumerate(schedule):
|
|
response, wall = diagnostic.request(dict(action='profile', request_id=f'profile-{index:02d}/' + entry['fixture_id'],
|
|
fixture_id=entry['fixture_id']))
|
|
record = dict(temperature=temperature, **compact_response(response, wall), cprofile=response['cprofile'])
|
|
save(diagnostic_dir / f'{index:02d}-{entry["fixture_id"]}.json', record)
|
|
diagnostics.append(dict(temperature=temperature, fixture_id=entry['fixture_id'],
|
|
profile_file=f'diagnostic/{index:02d}-{entry["fixture_id"]}.json',
|
|
self_time_categories_seconds=response['cprofile']['self_time_categories_seconds']))
|
|
diagnostic_exit = diagnostic.close()
|
|
warm = [row for row in resident if row['round_index'] > 0]
|
|
summary = dict(
|
|
cold_U_subprocess_wall_seconds=distribution([row['process']['process_wall_seconds'] for row in cold]),
|
|
cold_U_API_seconds=distribution([row['original_api_seconds'] for row in cold]),
|
|
cold_worker_startup_wall_seconds=distribution([row['process']['startup_wall_seconds'] for row in cold]),
|
|
cold_input_load_verify_seconds=distribution([row['input_load_verify_seconds'] for row in cold]),
|
|
resident_warm_U_API_seconds=distribution([row['producer']['original_api_seconds'] for row in warm]),
|
|
resident_warm_verify_API_seconds=distribution([row['verifier']['original_api_seconds'] for row in warm]),
|
|
resident_warm_pair_roundtrip_seconds=distribution([row['pair_parent_wall_seconds'] for row in warm]),
|
|
original_full_E_warm_API_seconds=distribution([row['original_api_seconds'] for row in epoch_resident if row['round_index'] == 1]),
|
|
original_full_E_warm_roundtrip_seconds=distribution([row['parent_roundtrip_seconds'] for row in epoch_resident if row['round_index'] == 1]),
|
|
original_full_E_cold_first_API_seconds=epoch_cold['original_api_seconds'],
|
|
original_full_E_cold_first_process_wall_seconds=epoch_cold['process']['process_wall_seconds'],
|
|
original_full_AQ_subprocess_wall_seconds=distribution([row['subprocess_wall_seconds'] for row in full]))
|
|
for path, pin in original_pins.items():
|
|
check(path, pin)
|
|
verify_suite(fixtures, expected)
|
|
check(Path(__file__).absolute(), own)
|
|
report = dict(schema=SCHEMA, complete=True, created_utc=datetime.now(timezone.utc).isoformat(),
|
|
registration_sha256=REG_SHA, original_source_commit=COMMIT, fixture_manifest_sha256=expected,
|
|
fixtures=14, resident_passes=repeats, warm_passes_in_summary=[1, 2], cold=cold, resident=resident,
|
|
full_E=dict(cold_fixed_first=epoch_cold, resident=epoch_resident, resident_passes=2,
|
|
historical_process_seconds={entry['fixture_id']: entry['historical_process_seconds']['update'] for entry in manifest['fixtures']}),
|
|
full_AQ=full, diagnostic=diagnostics, process_exits=dict(producer=producer_exit, verifier=verifier_exit,
|
|
profiler=diagnostic_exit, full_E_resident=epoch_exit, full_E_cold_first=epoch_cold_exit),
|
|
summary=summary, total_wall_seconds=time.perf_counter() - started,
|
|
limitations=['Fixed inputs replayed; no dependency is advanced and no produced weight is installed.',
|
|
'Resident original U verification omits full AQ checkpoint/graph hashing and cannot replace that validation in production.',
|
|
'cProfile is diagnostic only, with overhead; its cumulative function times overlap.',
|
|
'Full E resident retains every original validation check; stage times are not a measured complete brain/E/AQ pipeline.',
|
|
'Cold denotes a fresh Python process, not a flushed operating-system page cache.',
|
|
'CPU workload conditions are observed by the caller; these lanes execute serially.'],
|
|
original_files_modified=False, brain_executed=False, training_executed=False, weights_installed=False,
|
|
all_full_array_reference_comparisons_passed=True, all_original_receipts_exact=True,
|
|
independent_resident_verifier_process=True, all_child_exits_observed_zero=True,
|
|
full_AQ_validates_actual_second_pass_E_outputs=True,
|
|
files={str(path.relative_to(output)): identity(path) for path in sorted(output.rglob('*')) if path.is_file()})
|
|
save(output / 'report.json', report)
|
|
print(canonical(dict(complete=True, report=str(output / 'report.json'), summary=summary)), flush=True)
|
|
return report
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--fixtures', type=Path, required=True)
|
|
parser.add_argument('--fixture-manifest-sha256', required=True)
|
|
parser.add_argument('--output', type=Path)
|
|
parser.add_argument('--repeats', type=int, choices=(REPEATS,), default=REPEATS)
|
|
parser.add_argument('--worker', choices=('producer', 'verifier', 'epoch'), help=argparse.SUPPRESS)
|
|
parser.add_argument('--worker-output-root', type=Path, help=argparse.SUPPRESS)
|
|
args = parser.parse_args()
|
|
if args.worker:
|
|
require(args.output is None, 'Worker writes only its parent-controlled protocol streams')
|
|
worker(unlinked(args.fixtures), args.fixture_manifest_sha256, args.worker,
|
|
None if args.worker_output_root is None else unlinked(args.worker_output_root))
|
|
else:
|
|
require(args.output is not None, '--output is required')
|
|
try:
|
|
profile(args.fixtures, args.fixture_manifest_sha256, args.output, args.repeats)
|
|
finally:
|
|
cleanup_children()
|