From 17929b09fb41b8cdfab0a2a9e7a5c85ceccd8930 Mon Sep 17 00:00:00 2001 From: Digital Drosophila Archive Date: Thu, 1 Oct 2026 15:01:44 +0800 Subject: [PATCH] Profile unchanged original update pipelines with independent full verification --- .../profile_original.py | 770 ++++++++++++++++++ .../run_job.py | 57 +- .../test_profile_original.py | 100 +++ .../timeout_fixture.py | 8 + 4 files changed, 927 insertions(+), 8 deletions(-) create mode 100644 models/male-cns-lif/conditioning-pipeline-acceleration-v1/profile_original.py create mode 100644 models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_profile_original.py create mode 100644 models/male-cns-lif/conditioning-pipeline-acceleration-v1/timeout_fixture.py diff --git a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/profile_original.py b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/profile_original.py new file mode 100644 index 0000000..651be14 --- /dev/null +++ b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/profile_original.py @@ -0,0 +1,770 @@ +"""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() diff --git a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/run_job.py b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/run_job.py index b455b5c..8aeda28 100644 --- a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/run_job.py +++ b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/run_job.py @@ -6,16 +6,36 @@ import os from pathlib import Path import re import shutil +import signal import subprocess import time ROOT = Path('/home/qishizhe/phase4ar-pipeline-acceleration-20261001') + +def output_bytes(): + total = 0 + for f in ROOT.rglob('*'): + try: + if f.is_file() and not f.is_symlink(): total += f.stat().st_size + except FileNotFoundError: + pass # Concurrent compiler temporary file removal, not a payload. + return total + +def stop_group(proc): + try: os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: pass + try: proc.wait(timeout=5) + except subprocess.TimeoutExpired: pass + try: os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: pass + return proc.wait() + p = argparse.ArgumentParser() p.add_argument('--label', required=True) p.add_argument('--timeout', type=int, default=900) p.add_argument('command', nargs=argparse.REMAINDER) a = p.parse_args() -assert re.fullmatch(r'[a-z0-9-]+', a.label) and a.command and 0 < a.timeout <= 1800 +assert re.fullmatch(r'[a-z0-9-]+', a.label) and a.command and 0 < a.timeout <= 900 assert ROOT.is_dir() and not ROOT.is_symlink() assert shutil.disk_usage(ROOT).free >= 100 * (1 << 30) out = ROOT / 'jobs' @@ -25,15 +45,36 @@ record = dict(label=a.label, command=a.command, scope='independent engineering r started_utc=datetime.now(timezone.utc).isoformat(), scientific_AR_resumed=False) with (out / (a.label + '.start.json')).open('x') as f: json.dump(record, f, indent=2) with (out / (a.label + '.stdout.log')).open('x') as stdout, (out / (a.label + '.stderr.log')).open('x') as stderr: + proc = None try: - proc = subprocess.run(a.command, stdout=stdout, stderr=stderr, timeout=a.timeout, - env={**os.environ, 'PYTHONDONTWRITEBYTECODE':'1'}) - record.update(exit_code=proc.returncode, observed_exit=True, timed_out=False) - except subprocess.TimeoutExpired: - record.update(exit_code=None, observed_exit=False, timed_out=True) + # All children belong to a dedicated group created by THIS invocation. + # On stop, terminate this group only, never unrelated services/science. + proc = subprocess.Popen(a.command, stdout=stdout, stderr=stderr, start_new_session=True, + env={**os.environ, 'PYTHONDONTWRITEBYTECODE':'1'}) + record.update(pid=proc.pid, process_group=proc.pid) + stop_reason = None + while True: + try: + proc.wait(timeout=5) + break + except subprocess.TimeoutExpired: + if time.perf_counter() - start > a.timeout: stop_reason = 'wall_timeout' + elif shutil.disk_usage(ROOT).free < 100 * (1 << 30): stop_reason = 'minimum_disk_reserve' + elif output_bytes() > 2 * (1 << 30): + stop_reason = 'output_cap' + if stop_reason: + break + if not stop_reason and output_bytes() > 2 * (1 << 30): stop_reason = 'output_cap' + if stop_reason or proc.returncode != 0: stop_group(proc) + record.update(exit_code=proc.returncode, observed_exit=True, + timed_out=stop_reason == 'wall_timeout', stop_reason=stop_reason) except OSError as error: - record.update(exit_code=None, observed_exit=False, timed_out=False, launch_error=str(error)) + rc = None if proc is None else stop_group(proc) + record.update(exit_code=rc, observed_exit=proc is not None, timed_out=False, launch_error=str(error)) + except BaseException: + if proc is not None: stop_group(proc) + raise record.update(wall_seconds=time.perf_counter()-start, finished_utc=datetime.now(timezone.utc).isoformat()) with (out / (a.label + '.exit.json')).open('x') as f: json.dump(record, f, indent=2) print(json.dumps(record), flush=True) -raise SystemExit(record['exit_code'] if record['observed_exit'] else 124) +raise SystemExit(record['exit_code'] if record['observed_exit'] and not record.get('stop_reason') else 124) diff --git a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_profile_original.py b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_profile_original.py new file mode 100644 index 0000000..45f29cb --- /dev/null +++ b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/test_profile_original.py @@ -0,0 +1,100 @@ +"""Authority, full-byte IPC and cleanup tests without original production runs.""" +import base64 +import importlib.util +from pathlib import Path +import tempfile +import unittest + +import numpy as np + + +SCRIPT = Path(__file__).with_name('profile_original.py') +SPEC = importlib.util.spec_from_file_location('test_original_profile_module', SCRIPT) +P = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(P) + + +class BoundaryTests(unittest.TestCase): + def test_array_packet_uses_raw_signed_zero_bytes(self): + reference = np.asarray([0., 1.], dtype=np.float32) + P.compare_packet(P.pack_array(reference), reference) + changed = np.asarray([-0., 1.], dtype=np.float32) + with self.assertRaisesRegex(ValueError, 'full array bytes'): + P.compare_packet(P.pack_array(changed), reference) + + def test_packet_dtype_and_shape_are_not_repaired(self): + reference = np.asarray([1.], dtype=np.float32) + for array in (reference.astype(np.float64), reference.reshape(1, 1)): + with self.assertRaisesRegex(ValueError, 'identity'): + P.compare_packet(P.pack_array(array), reference) + + def test_response_chain_binds_all_output_bytes_and_role(self): + response = dict(role='producer', sequence=1, previous_response_sha256=None, complete=True, + arrays={'delta_after': P.pack_array(np.zeros(3, np.float32))}) + response['response_sha256'] = P.payload_sha(response) + self.assertEqual(P.verify_response(response, None, 1, 'producer'), response['response_sha256']) + with self.assertRaisesRegex(ValueError, 'chain/role'): + P.verify_response(response, None, 1, 'verifier') + response['arrays']['delta_after']['data_base64'] = base64.b64encode(b'bad').decode() + with self.assertRaisesRegex(ValueError, 'payload hash'): + P.verify_response(response, None, 1, 'producer') + + def test_response_chain_reordering_rejected(self): + response = dict(role='producer', sequence=2, previous_response_sha256='a' * 64, complete=True) + response['response_sha256'] = P.payload_sha(response) + with self.assertRaisesRegex(ValueError, 'chain/role'): + P.verify_response(response, 'b' * 64, 2, 'producer') + + def test_manifest_paths_cannot_escape(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary).resolve() + for name in ('../other', '/absolute', 'foo\\bar', 'x/../bar', 'C:/windows'): + with self.assertRaises(ValueError): + P.contained(root, name) + self.assertEqual(P.contained(root, 'valid/input.npz'), root / 'valid/input.npz') + + def test_output_writes_are_exclusive(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary).resolve() / 'result.json' + P.save(path, {'complete': True}) + original = path.read_bytes() + with self.assertRaises(FileExistsError): + P.save(path, {'complete': False}) + self.assertEqual(path.read_bytes(), original) + + def test_mismatching_source_pin_rejected(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary).resolve() / 'source.py' + path.write_text('raise AssertionError("must not run")', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'identity'): + P.load_source('untrusted_original_profile_test', path, '0' * 64) + + def test_profile_categories_do_not_use_cumulative_totals(self): + self.assertEqual(P.profile_category('/usr/lib/python3.10/ast.py', 'dump'), 'source_parse_compile_load') + self.assertEqual(P.profile_category('/frozen/phase4j.py::trace-expression-only', ''), + 'trace_expression_and_tensor_ops') + self.assertEqual(P.profile_category('/frozen/bridge.py', 'trace_from_events'), 'trace_apply_python_self') + + +class CleanupTests(unittest.TestCase): + def test_cleanup_attempts_every_owned_child_after_failure(self): + calls = [] + class Good: + def abort(self): + calls.append('good') + class Bad: + def abort(self): + calls.append('bad') + raise RuntimeError('simulated abort error') + original = list(P.CHILDREN) + try: + P.CHILDREN[:] = [Good(), Bad()] + with self.assertRaisesRegex(ValueError, 'cleanup errors'): + P.cleanup_children() + self.assertEqual(calls, ['bad', 'good']) + finally: + P.CHILDREN[:] = original + + +if __name__ == '__main__': + unittest.main() diff --git a/models/male-cns-lif/conditioning-pipeline-acceleration-v1/timeout_fixture.py b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/timeout_fixture.py new file mode 100644 index 0000000..cbeae64 --- /dev/null +++ b/models/male-cns-lif/conditioning-pipeline-acceleration-v1/timeout_fixture.py @@ -0,0 +1,8 @@ +"""Safety self-test only: a child/grandchild sleep tree to terminate by own group.""" +import subprocess +import sys +import time + +child = subprocess.Popen([sys.executable, '-B', '-c', 'import time; time.sleep(90)']) +print(child.pid, flush=True) +time.sleep(90)