diff --git a/tools/conditioning-perf-lab-v1/README.md b/tools/conditioning-perf-lab-v1/README.md new file mode 100644 index 0000000..bdef940 --- /dev/null +++ b/tools/conditioning-perf-lab-v1/README.md @@ -0,0 +1,48 @@ +# Independent CPU residency performance probe + +This is a bounded engineering experiment authorized after the user asked to try +reducing repeated process startup. It does not replace or modify registered AR, +its timing/seed/dynamics, any frozen model, or any existing scientific artifact. +No GPU worker or new brain simulation is launched, and AR interim learning +outputs are not inspected. Input selection uses the two named, already sealed +AQ engineering fixtures, not outcomes from the ongoing experiment. + +## Fixed experiment + +Serial sequence: M, N, M, N_D0, N_frozen, M_D0, M. M is the mechanical fresh +parent; N is the natural light fresh engineering trial, whose actual prior +overlay comes from an AP learned snapshot. The three explicitly derived command +controls change only D=0/frozen over the same sealed events. These are not new +conditioning trials and do not alter the underlying files or install weights. + +Compare fresh original epoch postprocessing against the exact same function +called repeatedly in a dedicated worker. Preserve all original hashes and +checkpoint validation; do not cache mutable neural state or delta. Every output +is independently checked by the original fresh-process CPU validator. A second, +separate resident validator is a clearly identified prototype: it has a new +schema/source identity, caches only SHA-checked module identity, and preserves +the original numerical and input-checking body. + +Require exact four-file SHA identity, raw delta dtype/shape/byte identity, +complete update-receipt/trace-digest identity, and identical scientific verifier +fields. Compare mechanical and natural cases back to their original AQ outputs. +Check D0/frozen byte identity, and source/consumed-input hashes before and after. +Each task ACK is distinct from a true worker process exit. Both resident workers +must exit zero as observed by their parent; every fresh validator must exit zero. + +One active CPU task at a time, one numerical thread, low scheduling priority, +CUDA hidden and uninitialized, at most seven rows, 180-second child-response +limit and 900-second experiment limit. Fail closed, preserve partial outputs, +no automatic retry. New output root must be an empty temporary directory outside +the registered experiment/model trees. No full-history rescan is claimed. + +## Interpretation limits + +Report cold and warm timings separately. Fresh reference and resident treatment +use the same unchanged inputs, original function body and complete validation. +The comparison excludes GPU/brain work, AP outer wrapper and AR ledger overhead. +It cannot establish whole-run speedup, GPU correctness, sequence generality, +learning success or readiness to replace AR. Nonzero prior delta is included, +but only two source trajectories and one compartment are covered. Broadening to +different compartments or a persistent GPU runtime requires a new independent +equivalence test and authorization before replacing a registered workflow. diff --git a/tools/conditioning-perf-lab-v1/perf_probe.py b/tools/conditioning-perf-lab-v1/perf_probe.py new file mode 100644 index 0000000..318ded1 --- /dev/null +++ b/tools/conditioning-perf-lab-v1/perf_probe.py @@ -0,0 +1,370 @@ +"""Bounded CPU-only residency experiment; never an AR runtime replacement.""" +from __future__ import annotations +import argparse +import contextlib +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import selectors +import shutil +import statistics +import subprocess +import sys +import time +import traceback + +ROOT = Path('/data/digital-drosophila') +LAB = ROOT/'models/male-cns-lif/olfactory-conditioning-lab-v1' +AQ = ROOT/'experiments/phase4aq-conditioning-execution' +VALIDATOR = ROOT/'models/male-cns-lif/conditioning-execution-v1/validate_update_cpu.py' +E_SHA = '757935a4815d9f625640df685184ed3fe44a8845323562796da7aa43cbb194e1' +V_SHA = '44e95eca9e39e9cd25a1d5d2d46b2cea76e839143865a196f865a61fac700b9f' +MANIFEST_SHA = '2dc0feca75f718f2ee0d7c9240da503221823573fd25f1344d2fd97d97eb2db7' +FIXTURES = { + 'M': ('mechanical--fresh-parent', '4616a42c255ab839346368aa0e5e4306bb3878e55fcc95b176f058b1993c9b0c'), + 'N': ('performance--natural-light-fresh', '785e3d26d06b47efe7f054883d38fe2e13711195359bf9f4b03aa88887fbfb37'), +} +SEQUENCE = ('M', 'N', 'M', 'N_D0', 'N_frozen', 'M_D0', 'M') +UPDATE_FILES = ('delta-after.npz', 'overlay-manifest.json', 'update-receipt.json', 'chain-receipt.json') +TIMEOUT = 180 + + +def require(ok, message): + if not ok: + raise ValueError(message) + + +def digest(path): + h = hashlib.sha256() + with Path(path).open('rb') as stream: + for chunk in iter(lambda: stream.read(8 << 20), b''): + h.update(chunk) + return h.hexdigest() + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False) + + +def read(path): + return json.loads(Path(path).read_text(encoding='utf-8')) + + +def save(path, value): + with Path(path).open('x', encoding='utf-8', newline='\n') as stream: + json.dump(value, stream, sort_keys=True, indent=2, allow_nan=False) + stream.write('\n') + + +def safe(path): + require('..' not in Path(path).parts, 'Parent traversal forbidden') + path = Path(path).absolute() + require(not any(p.is_symlink() for p in (path, *path.parents)), 'Symlink path forbidden') + return path.resolve(strict=False) + + +def pinned_load(name, path, sha): + path = safe(path) + raw = 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 worker(role, mode, plan_file, plan_sha): + require(sys.dont_write_bytecode and os.environ.get('CUDA_VISIBLE_DEVICES') == '', 'CPU-only -B worker required') + require(digest(plan_file) == plan_sha, 'External plan SHA mismatch') + plan = read(plan_file) + out = safe(plan['output_root']) + pins = plan['prototype_sources'] + require(all(digest(p) == s for p, s in pins.items()), 'Prototype source changed') + module = None + seq = 0 + print(canonical({'ready': True, 'pid': os.getpid(), 'role': role}), flush=True) + for line in sys.stdin: + job = json.loads(line) + require(set(job) == {'sequence', 'authority', 'authority_sha256', 'output', 'update_dir', 'chain_sha256'}, 'Wrong request fields') + require(seq < len(SEQUENCE) and job['sequence'] == seq and + job['authority'] in plan['authorities'].values() and + plan['input_sha256'][job['authority']] == job['authority_sha256'] and + digest(job['authority']) == job['authority_sha256'], 'Sequence/authority differs') + output = safe(job['output']) + require(output.is_relative_to(out) and not output.exists() and output.parent.is_dir(), 'Output must be exclusive and within experiment') + started = time.perf_counter() + with contextlib.redirect_stdout(sys.stderr): + if module is None: + if role == 'update': + module = pinned_load('_perf_E', LAB/'conditioning_epoch.py', E_SHA) + elif role == 'verify-reference': + module = pinned_load('_perf_original_V', VALIDATOR, V_SHA) + else: + source = Path(__file__).parent/'resident_validator.py' + module = pinned_load('_perf_resident_V', source, pins[str(source)]) + if role == 'update': + module.postprocess_epoch(job['authority'], job['authority_sha256'], output) + hashes = {name: digest(output/name) for name in UPDATE_FILES} + else: + module.validate(job['authority'], job['authority_sha256'], job['update_dir'], job['chain_sha256'], output) + hashes = {'receipt': digest(output)} + import torch + require(os.environ['CUDA_VISIBLE_DEVICES'] == '' and not torch.cuda.is_initialized(), 'CUDA use forbidden') + require('runtime_light' not in sys.modules, 'Ordinary runtime module collision') + require(all(digest(p) == s for p, s in pins.items()), 'Prototype source changed during task') + print(canonical({'ack': True, 'sequence': seq, 'pid': os.getpid(), 'role': role, + 'request_sha256': hashlib.sha256(canonical(job).encode()).hexdigest(), + 'output_sha256': hashes, 'task_seconds': time.perf_counter()-started, + 'CUDA_initialized': False, 'task_ack_is_not_process_exit': True}), flush=True) + seq += 1 + if mode == 'once': + break + + +class Process: + def __init__(self, role, mode, plan, root, label, deadline): + self.role, self.mode, self.seq = role, mode, 0 + self.deadline = deadline + self.label, self.root, self.records = label, root, [] + self.stderr_path = root/'logs'/(label+'.stderr.log') + self.stderr = self.stderr_path.open('xb') + env = dict(os.environ, CUDA_VISIBLE_DEVICES='', PYTHONDONTWRITEBYTECODE='1', + OMP_NUM_THREADS='1', MKL_NUM_THREADS='1', OPENBLAS_NUM_THREADS='1') + argv = [sys.executable, '-B', str(Path(__file__).resolve()), 'worker', '--role', role, + '--mode', mode, '--plan', str(plan), '--plan-sha', digest(plan)] + self.started = time.perf_counter(); self.argv = argv + self.selector = selectors.DefaultSelector() + self.p = None + try: + self.remaining() + self.p = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=self.stderr, text=True, bufsize=1, env=env) + self.selector.register(self.p.stdout, selectors.EVENT_READ) + self.ready = self.line() + require(self.ready == {'ready': True, 'pid': self.p.pid, 'role': role}, 'Wrong worker ready record') + self.startup_seconds = time.perf_counter()-self.started + except BaseException: + self.abort('Failed startup/ready') + raise + + def remaining(self): + remaining = self.deadline-time.perf_counter() + require(remaining > 0, 'Probe absolute wall deadline exceeded') + return min(TIMEOUT, remaining) + + def abort(self, reason): + if self.p is not None: + if self.p.poll() is None: + self.p.terminate() + try: self.p.wait(timeout=10) + except subprocess.TimeoutExpired: self.p.kill(); self.p.wait(timeout=10) + for stream in (self.p.stdin, self.p.stdout): + if stream is not None: stream.close() + path = self.root/'processes'/(self.label+'.aborted.json') + if not path.exists(): + save(path, dict(pid=self.p.pid, argv=self.argv, actual_parent_observed_exit=self.p.returncode, + reason=reason, partial_preserved=True, benchmark_failed=True, no_retry=True)) + self.stderr.close(); self.selector.close() + + def line(self): + require(self.selector.select(self.remaining()), 'Worker response timeout; no retry') + raw = self.p.stdout.readline() + require(raw, 'Worker EOF before expected response; inspect stderr') + return json.loads(raw) + + def request(self, authority, output, update_dir=None): + request_started = time.perf_counter() + job = dict(sequence=self.seq, authority=str(authority), authority_sha256=digest(authority), + output=str(output), update_dir=None if update_dir is None else str(update_dir), + chain_sha256=None if update_dir is None else digest(update_dir/'chain-receipt.json')) + save(self.root/'requests'/(self.label+f'-{self.seq:02d}.json'), job) + started = time.perf_counter() + self.p.stdin.write(canonical(job)+'\n'); self.p.stdin.flush() + ack = self.line() + require(ack.get('ack') is True and ack['sequence'] == self.seq and ack['pid'] == self.p.pid and + ack['role'] == self.role and ack['request_sha256'] == hashlib.sha256(canonical(job).encode()).hexdigest() and + ack['CUDA_initialized'] is False and ack['task_ack_is_not_process_exit'] is True, 'Worker acknowledgment differs') + hashes = ({name: digest(output/name) for name in UPDATE_FILES} if self.role == 'update' else {'receipt': digest(output)}) + require(hashes == ack['output_sha256'], 'Worker output changed or acknowledgment differs') + record = dict(ack=ack, rpc_parent_seconds=time.perf_counter()-started) + save(self.root/'responses'/(self.label+f'-{self.seq:02d}.json'), record) + record['parent_request_seconds'] = time.perf_counter()-request_started + self.records.append(record); self.seq += 1 + return record + + def close(self): + self.p.stdin.close() + rc = self.p.wait(timeout=self.remaining()) + elapsed = time.perf_counter()-self.started + trailing = self.p.stdout.read() + self.p.stdout.close(); self.stderr.close(); self.selector.close() + receipt = dict(pid=self.p.pid, argv=self.argv, role=self.role, mode=self.mode, + actual_parent_observed_exit=rc, tasks_acknowledged=self.seq, startup_seconds=self.startup_seconds, + lifetime_parent_seconds=elapsed, stderr_sha256=digest(self.stderr_path), + task_ACKs_are_not_individual_OS_exits=True) + save(self.root/'processes'/(self.label+'.json'), receipt) + require(rc == 0 and not trailing, 'Worker actual nonzero exit or unsolicited output') + return receipt + + +def compare_updates(a, b): + import numpy as np + require(set(x.name for x in a.iterdir()) == set(UPDATE_FILES) == set(x.name for x in b.iterdir()), 'Wrong update inventory') + for name in UPDATE_FILES: + require(digest(a/name) == digest(b/name), 'Exact output file differs: '+name) + with np.load(a/'delta-after.npz', allow_pickle=False) as x, np.load(b/'delta-after.npz', allow_pickle=False) as y: + require(x.files == y.files, 'Array schema differs') + for key in x.files: + require(x[key].dtype == y[key].dtype and x[key].shape == y[key].shape and x[key].tobytes() == y[key].tobytes(), 'Raw array differs') + return {'all_four_files_SHA_identical': True, 'arrays_bitwise_identical': True, + 'full_update_receipt_and_trace_digests_identical': True} + + +def compare_validation(reference, resident): + a, b = read(reference), read(resident) + require(a.pop('schema') == 'conditioning-update-cpu-validation-v1', 'Wrong original verifier schema') + require(b.pop('schema') == 'conditioning-resident-cpu-validation-prototype-v1', 'Wrong prototype verifier schema') + original = a.pop('validator_source') + require(original == {'path': str(VALIDATOR), 'sha256': V_SHA}, 'Wrong original validator identity') + require(b.pop('original_validator_source') == original and b.pop('resident_prototype') is True, 'Missing prototype original-source provenance') + require(b.pop('resident_cache_scope') == 'SHA-verified imported modules only; no input/hash/state/delta/checkpoint caching', 'Wrong cache scope') + identity = b.pop('validator_source') + require(identity == {'path': str(Path(__file__).resolve().parent/'resident_validator.py'), + 'sha256': digest(Path(__file__).parent/'resident_validator.py')}, 'Wrong resident source identity') + require(a == b, 'Scientific verifier receipt differs beyond enumerated provenance fields') + return {'all_scientific_fields_identical': True, 'explicit_differing_fields': + ['schema', 'validator_source', 'original_validator_source', 'resident_prototype', 'resident_cache_scope']} + + +def prepare(root): + require(sys.dont_write_bytecode and root.is_dir() and not list(root.iterdir()), 'New empty output root and -B required') + require(not root.is_relative_to(ROOT/'experiments') and not root.is_relative_to(ROOT/'models'), 'Prototype may not write experimental/model trees') + require(shutil.disk_usage(root).free > (10 << 30), 'Insufficient temporary free space') + for name in ('inputs', 'updates', 'validation', 'logs', 'requests', 'responses', 'processes'): + (root/name).mkdir() + require(digest(AQ/'manifest.json') == MANIFEST_SHA, 'AQ sealed manifest differs') + manifest = {x['path']: x for x in read(AQ/'manifest.json')['files']} + pins = {str(AQ/'manifest.json'): MANIFEST_SHA, str(LAB/'conditioning_epoch.py'): E_SHA, str(VALIDATOR): V_SHA} + authorities = {} + for label, (name, expected) in FIXTURES.items(): + authority = AQ/'authorities'/(name+'.epoch.json') + require(manifest[authority.relative_to(AQ).as_posix()]['sha256'] == expected and digest(authority) == expected, 'AQ fixture authority seal differs') + authorities[label] = authority + a = read(authority); pins[str(authority)] = expected + pins.update(a['frozen_inputs']) + pins.update({x['path']: x['sha256'] for x in a['files'].values()}) + checkpoint = Path(a['checkpoint_dir']); cp = checkpoint/'checkpoint-manifest.json' + require(manifest[cp.relative_to(AQ).as_posix()]['sha256'] == a['checkpoint_manifest_sha256'], 'Checkpoint not in fixed AQ seal') + pins[str(cp)] = a['checkpoint_manifest_sha256'] + for f in checkpoint.iterdir(): + require(f.is_file() and not f.is_symlink(), 'Unexpected checkpoint entry') + pins[str(f)] = manifest[f.relative_to(AQ).as_posix()]['sha256'] + for filename in UPDATE_FILES: + f = AQ/'updates'/name/filename + pins[str(f)] = manifest[f.relative_to(AQ).as_posix()]['sha256'] + for label in ('N_D0', 'N_frozen', 'M_D0'): + a = read(authorities[label[0]]) + a['epoch_id'] = 'engineering-residency-control-'+label + if label.endswith('_D0'): + a['command']['d_command'] = 0.0 + else: + a['command']['frozen'] = True + path = root/'inputs'/(label+'.authority.json'); save(path, a) + authorities[label] = path; pins[str(path)] = digest(path) + require(all(digest(p) == sha for p, sha in pins.items()), 'Frozen fixture input bytes differ') + sources = {str(p.resolve()): digest(p) for p in Path(__file__).parent.glob('*.py')} + plan = dict(schema='conditioning-cpu-residency-probe-v1', output_root=str(root), sequence=list(SEQUENCE), + authorities={k: str(v) for k, v in authorities.items()}, input_sha256=pins, prototype_sources=sources, + fixture_selection='Two predeclared complete frozen AQ engineering fixtures; never AR interim results', + derived_controls='Only D=0/frozen command variants over fixed sealed events; not new brain runs', + threads=1, CPU_only=True, no_brain_execution=True, no_weight_installation=True, + current_AR_untouched=True, no_automatic_retry=True, max_wall_seconds=900, + benchmark_scope='CPU epoch postprocessing and verification only; excludes AP wrapper/ledger and all GPU/brain work', + acceptance='Exact four update files, bitwise delta, full trace/receipt identity, fresh original verifier exit0, separate resident verifier identity') + save(root/'benchmark-plan.json', plan) + return plan + + +def run(root): + require(root.is_dir() and not list(root.iterdir()), 'New empty output directory required') + started = time.perf_counter(); deadline = started+900 + active = [] + try: + plan = prepare(root); planfile = root/'benchmark-plan.json' + # Yield to the live experiment; only one worker computes at a time. + os.nice(19) + e = Process('update', 'server', planfile, root, 'resident-E', deadline); active.append(e) + v = Process('verify-resident', 'server', planfile, root, 'resident-V', deadline); active.append(v) + require(e.p.pid != v.p.pid, 'Independent workers required') + rows = [] + for number, label in enumerate(SEQUENCE): + require(time.perf_counter()-started < 900, 'Probe wall cap exceeded') + authority = Path(plan['authorities'][label]); prefix = f'{number:02d}-{label}' + refout, resout = (root/'updates'/(prefix+'-'+x) for x in ('fresh', 'resident')) + reference = Process('update', 'once', planfile, root, prefix+'-fresh-E', deadline); active.append(reference) + reference.request(authority, refout); ereceipt = reference.close(); active.remove(reference) + er = e.request(authority, resout) + equality = compare_updates(refout, resout) + if label in FIXTURES: + compare_updates(AQ/'updates'/FIXTURES[label][0], resout) + freshv, residentv = (root/'validation'/(prefix+'-'+x+'.json') for x in ('fresh', 'resident')) + validator = Process('verify-reference', 'once', planfile, root, prefix+'-fresh-V', deadline); active.append(validator) + validator.request(authority, freshv, resout); vreceipt = validator.close(); active.remove(validator) + vr = v.request(authority, residentv, resout) + vequality = compare_validation(freshv, residentv) + if label.endswith('_D0') or label.endswith('_frozen'): + import numpy as np + before = Path(read(authority)['files']['overlay_before']['path']) + with np.load(before, allow_pickle=False) as a, np.load(resout/'delta-after.npz', allow_pickle=False) as b: + require(a['plastic_delta_mv'].tobytes() == b['plastic_delta_mv'].tobytes(), 'D0/frozen failed bitwise identity') + row = dict(label=label, position=number, fresh_E_seconds=ereceipt['lifetime_parent_seconds'], + resident_E_seconds=er['parent_request_seconds'], fresh_V_seconds=vreceipt['lifetime_parent_seconds'], + resident_V_seconds=vr['parent_request_seconds'], update_comparison=equality, validator_comparison=vequality, + original_fresh_validator_actual_exit=vreceipt['actual_parent_observed_exit']) + rows.append(row); save(root/(prefix+'-comparison.json'), row) + print(canonical({'completed': number+1, 'total': len(SEQUENCE), 'label': label, 'exact_equal': True}), flush=True) + ep, vp = e.close(), v.close(); active.clear() + require(all(digest(p) == sha for p, sha in plan['input_sha256'].items()), 'Consumed frozen inputs changed') + require(all(digest(p) == sha for p, sha in plan['prototype_sources'].items()), 'Prototype changed') + require(time.perf_counter() < deadline, 'Probe absolute wall deadline exceeded before PASS') + warm = rows[1:] + timing = {} + for field in ('fresh_E_seconds','resident_E_seconds','fresh_V_seconds','resident_V_seconds'): + timing[field] = dict(all_mean=statistics.mean(r[field] for r in rows), + warm_mean=statistics.mean(r[field] for r in warm), warm_median=statistics.median(r[field] for r in warm)) + timing['warm_combined_speedup_ratio'] = sum(r['fresh_E_seconds']+r['fresh_V_seconds'] for r in warm)/sum(r['resident_E_seconds']+r['resident_V_seconds'] for r in warm) + timing['warm_E_only_with_original_fresh_verifier_speedup_ratio'] = sum(r['fresh_E_seconds']+r['fresh_V_seconds'] for r in warm)/sum(r['resident_E_seconds']+r['fresh_V_seconds'] for r in warm) + report = dict(status='PASS-CPU-RESIDENCY-PROTOTYPE-ONLY', trials=rows, timing=timing, + resident_process_exits=[ep['actual_parent_observed_exit'], vp['actual_parent_observed_exit']], + source_and_consumed_input_SHA_unchanged=True, full_history_reaudit=False, + prototype_plan_sha256=digest(planfile), total_wall_seconds=time.perf_counter()-started, + timing_boundary='Parent request includes input/chain digest, request+response persistence and output verification. Fresh also includes spawn/ready/exit; resident lifetime/startup separately recorded.', + AR_modified=False, brain_run=False, GPU_benchmarked=False, static_hash_cache_removed=False, + learning_test=False, candidate_generated=False, end_to_end_AR_speedup_not_measured=True, + scope='7 fixed engineering replays; CPU postprocessing only. Not authorized or validated as AR replacement.') + save(root/'report.json', report) + files = {str(p.relative_to(root)): {'sha256': digest(p), 'bytes': p.stat().st_size} for p in root.rglob('*') if p.is_file()} + save(root/'manifest.json', {'files': files, 'manifest_self_excluded': True}) + print(canonical({'report': str(root/'report.json'), 'status': report['status'], 'timing': timing}), flush=True) + except BaseException as error: + for process in active: + process.abort(type(error).__name__+': '+str(error)) + if not (root/'stopped.json').exists(): + save(root/'stopped.json', dict(error_type=type(error).__name__, error=str(error), traceback=traceback.format_exc(), + partial_preserved=True, no_automatic_retry=True, registered_AR_not_touched=True)) + raise + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest='mode', required=True) + r = sub.add_parser('run'); r.add_argument('--output', required=True) + w = sub.add_parser('worker'); w.add_argument('--role', choices=('update','verify-reference','verify-resident'), required=True) + w.add_argument('--mode', dest='worker_mode', choices=('once','server'), required=True) + w.add_argument('--plan', required=True); w.add_argument('--plan-sha', required=True) + args = parser.parse_args() + if args.mode == 'run': run(safe(args.output)) + else: worker(args.role, args.worker_mode, args.plan, args.plan_sha) diff --git a/tools/conditioning-perf-lab-v1/resident_validator.py b/tools/conditioning-perf-lab-v1/resident_validator.py new file mode 100644 index 0000000..eea09a5 --- /dev/null +++ b/tools/conditioning-perf-lab-v1/resident_validator.py @@ -0,0 +1,228 @@ +"""CPU-only resident validation prototype; not the frozen AQ validator. + +All epoch computation and file checks remain the frozen validation body. Only +pinned source modules persist: no state, delta, input hash or checkpoint cache. +Use a dedicated python -B process with CUDA hidden. A failed request poisons +this worker; it never repairs module caches, retries, runs or changes a brain. +A caller must independently authenticate every request and actual process exit. +""" +from __future__ import annotations + +import argparse +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import sys + +import numpy as np + +FROZEN_MODEL_ROOT = Path('/data/digital-drosophila/models/male-cns-lif') +LAB = FROZEN_MODEL_ROOT/'olfactory-conditioning-lab-v1' +E_SHA = '757935a4815d9f625640df685184ed3fe44a8845323562796da7aa43cbb194e1' +Q_SHA = 'b88fd9a88d9f92d5c4d042b90f2129b015e8d4bd810c41351ee573241e2d4c2e' +U_SHA = '96efff678cff910654eb08e86f1a17789e4e5589e340345f419c7027a7d547e2' +CMP_SHA = '5a2403f11bc08d65a9d242ac2cf9686b850e0c3680d65659f785e9b6b50c0238' +FILES = ('chain-receipt.json','update-receipt.json','delta-after.npz','overlay-manifest.json') +ORIGINAL_VALIDATOR = FROZEN_MODEL_ROOT/'conditioning-execution-v1/validate_update_cpu.py' +ORIGINAL_VALIDATOR_SHA = '44e95eca9e39e9cd25a1d5d2d46b2cea76e839143865a196f865a61fac700b9f' +_LOAD_CACHE = {} +_RESIDENT_STATE = 'new' +_TORCH_MODULE = None +_SELF_SOURCE_SHA = None + + +def require(ok,message): + if not ok: raise ValueError(message) + + +def digest(path): + h=hashlib.sha256() + with Path(path).open('rb') as stream: + for b in iter(lambda:stream.read(8<<20),b''):h.update(b) + return h.hexdigest() + + +def same(a,b): return a.dtype==b.dtype and a.shape==b.shape and a.tobytes()==b.tobytes() + + +def unlinked(path): + path=Path(path).absolute() + require(not any(p.is_symlink() or (hasattr(p,'is_junction') and p.is_junction()) + for p in (path,*path.parents)), 'Linked receipt/update path forbidden') + return path + + +def load(name,path,expected): + # Fresh bytes on EVERY call, even when an import is already resident. + path=unlinked(path).resolve(strict=True);raw=path.read_bytes() + require(hashlib.sha256(raw).hexdigest()==expected,'Pinned source bytes changed') + if name in _LOAD_CACHE: + saved_path,saved_sha,module=_LOAD_CACHE[name] + require(saved_path==path and saved_sha==expected and sys.modules.get(name) is module and + Path(module.__file__).resolve()==path,'Resident source/module identity differs') + return module + require(name not in sys.modules,'Unverified source/module binding') + spec=importlib.util.spec_from_file_location(name,path);module=importlib.util.module_from_spec(spec) + sys.modules[name]=module + exec(compile(raw,str(path),'exec'),module.__dict__) + _LOAD_CACHE[name]=(path,expected,module) + return module + + +def chain_binding(chain,authority,authority_sha,pins,paths,checkpoint,hashes): + require(chain.get('schema')=='phase4an-epoch-postprocess-chain-v1' and chain.get('complete') is True, + 'Incomplete epoch update chain') + for flag in ('weights_installed','brain_executed','learning_pass_claimed','candidate_generated'): + require(chain.get(flag) is False,'Wrong chain scope: '+flag) + require(chain['coordinator_source_sha256']==E_SHA and chain['authority_sha256']==authority_sha and + chain['epoch_id']==authority['epoch_id'] and chain['command']==authority['command'] and + chain['diagnostic_mode']==authority['diagnostic_mode'],'Epoch authority/command differs') + require(chain['source_neural_checkpoint_dir']==str(checkpoint) and + chain['source_neural_checkpoint_manifest_sha256']==authority['checkpoint_manifest_sha256'] and + chain['input_files_sha256']=={str(p):v for p,v in pins.items()},'Actual checkpoint/input binding differs') + for field,role in (('source_state_file_sha256','state'),('source_spec_sha256','spec'), + ('source_requests_sha256','requests'),('before_overlay_file_sha256','overlay_before'), + ('before_overlay_manifest_sha256','overlay_manifest_before')): + require(chain[field]==pins[paths[role]],'Chain input differs: '+role) + expected={name:hashes[name] for name in FILES if name!='chain-receipt.json'} + require(chain['output_files_sha256']==expected and chain['after_overlay_file_sha256']==hashes['delta-after.npz'] and + chain['after_overlay_manifest_sha256']==hashes['overlay-manifest.json'],'Chain output differs') + + +def _validate_checked(epoch_authority,authority_sha,update_dir,chain_sha,receipt_file): + e=load('_conditioning_execution_E_validator',LAB/'conditioning_epoch.py',E_SHA) + load('engineering_compare',LAB/'engineering_compare.py',CMP_SHA) + q=load('_conditioning_execution_Q_validator',LAB/'conditioning_preflight_compare.py',Q_SHA) + value,paths,pins,checkpoint,authority_path=e._authority(epoch_authority,authority_sha) + c,r,u=e._production_dependencies(paths,pins) # The only R is C.R; no ordinary runtime loader. + directory=unlinked(update_dir).resolve(strict=True);out=unlinked(receipt_file) + protected={checkpoint,directory,*[p.parent for p in pins]} + require(out.parent.is_dir() and not out.exists() and not any(out.resolve().is_relative_to(p) for p in protected), + 'Receipt must be new and outside every frozen input/output directory') + require(set(p.name for p in directory.iterdir())==set(FILES) and + all((directory/n).is_file() and not (directory/n).is_symlink() for n in FILES),'Expected exact four update files') + hashes={n:digest(directory/n) for n in FILES} + require(e._sha(chain_sha) and hashes['chain-receipt.json']==chain_sha,'Chain differs from external SHA') + own=digest(__file__);chain=e._read(directory/'chain-receipt.json') + chain_binding(chain,value,authority_sha,pins,paths,checkpoint,hashes) + parent=c.load_checkpoint(checkpoint,value['checkpoint_manifest_sha256']) + command=value['command'];spec=e._read(paths['spec']) + r.validate_spec(spec,paths['spec'].parent) + require(parent['fresh'] is True and parent['global_tick_offset']==0 and parent['steps']==command['steps']==spec['steps'] and + parent['summary']['diagnostic_mode']==value['diagnostic_mode'],'Wrong actual epoch/state scope') + for role in ('spec','requests','overlay_before','overlay_manifest_before'): + require(parent['provenance']['input_sha256'].get(str(paths[role]))==pins[paths[role]],'Runtime input differs: '+role) + require(parent['manifest']['files_sha256']['state.npz']==pins[paths['state']],'State not sealed by this checkpoint') + _,before,before_meta,_=r.load_overlay(paths['overlay_before'],paths['overlay_manifest_before']) + _,after,after_meta,_=r.load_overlay(directory/'delta-after.npz',directory/'overlay-manifest.json') + graph=Path(spec['graph_dir']).resolve();apl=Path(spec['apl_bundle_dir']).resolve() + for location in [graph/n for n in e.GRAPH_FILES]+[apl/n for n in e.APL_FILES]+[Path(spec['coefficients_path']).resolve()]: + require(location in pins,'Missing complete graph/APL/coefficient pin') + require(paths['canonical_kc']==apl/'kc_globals.npy','Canonical KC path differs') + for role,prefix in (('mask','mask'),('base_candidate','base_candidate'),('am_overlay_source','am_overlay')): + require(Path(before_meta[prefix+'_path']).resolve()==paths[role] and before_meta[prefix+'_sha256']==pins[paths[role]],'Overlay input differs') + aj=Path(before_meta['aj_model_dir']).resolve() + for relative,sha in r.PINS.items():require(pins.get((aj.parent/relative).resolve())==sha,'Missing original dependency pin') + mapping=e._npz(paths['candidate_map'],('edge_id','source_neuron_idx','target_neuron_idx','target_compartment_verbatim')) + mask=np.load(paths['mask'],allow_pickle=False);kc=np.load(paths['kc'],allow_pickle=False) + canonical=np.load(paths['canonical_kc'],allow_pickle=False);base=np.load(paths['base_candidate'],allow_pickle=False) + e._vector(mask,'int64',14232,'mask');e._vector(kc,'int32',4064,'KC');e._vector(canonical,'int32',4064,'canonical KC') + e._vector(base,'float32',14232,'base') + require(np.all((mask>=0)&(mask<25582938)) and np.all(mask[1:]>mask[:-1]) and + np.all((kc>=0)&(kc<166700)) and len(np.unique(kc))==len(kc), + 'Candidate/KC range or uniqueness differs') + require(same(kc,canonical) and same(mask,mapping['edge_id']) and same(mask,before['edge_id']) and same(mask,after['edge_id']), + 'KC/candidate membership differs') + require(same(base,before['base_candidate_mv']) and same(base,after['base_candidate_mv']),'Original base changed') + maps={} + try: + for n,dtype,count in (('nodes','int32',166700),('edge_id','int64',25582938),('src','int32',25582938),('dst','int32',25582938)): + maps[n]=np.load(graph/(n+'.npy'),allow_pickle=False,mmap_mode='r');e._vector(maps[n],dtype,count,n) + require(np.array_equal(maps['nodes'],np.arange(166700,dtype=np.int32)) and same(maps['edge_id'][mask],mask),'Not canonical full graph') + require(same(mapping['source_neuron_idx'],maps['src'][mask]) and same(mapping['target_neuron_idx'],maps['dst'][mask]), + 'Candidate endpoints differ from actual connectome') + finally: + for a in maps.values():a._mmap.close() + state=e._npz(paths['state'],('step','neuron_idx','counts','v','g','ref')) + require(all(same(state[n],parent['state'][n]) for n in state),'Actual sealed event/state mismatch') + require(parent['overlay']['delta_raw_sha256']==r._raw(before['plastic_delta_mv']) and + chain['runtime_delta_before_raw_sha256']==r._raw(before['plastic_delta_mv']) and + chain['runtime_effective_weights_raw_sha256']==parent['summary']['effective_weights_raw_sha256'], 'Actual before installation differs') + update=e._read(directory/'update-receipt.json') + require(update['payload_sha256']==chain['update_receipt_payload_sha256'],'Update receipt chain differs') + u.verify_receipt(update,step=state['step'],neuron_idx=state['neuron_idx'],kc=kc,candidate_edge_id=mask, + candidate_sources=mapping['source_neuron_idx'],candidate_targets=mapping['target_neuron_idx'], + candidate_compartments=mapping['target_compartment_verbatim'],delta_before=before['plastic_delta_mv'],base_candidate=base, + steps=command['steps'],n_neurons=166700,compartment=command['compartment'],us_tick=command['us_tick'], + d_command=command['d_command'],frozen=command['frozen'],trace_backend=command['trace_backend'], + bridge_source=paths['eligibility_bridge_source'],phase4j_source=paths['phase4j_source']) + binding=q.audit_delta_binding(before['plastic_delta_mv'],after['plastic_delta_mv'],after['plastic_delta_mv'], + update_receipt=update,parent_snapshot_sha256=pins[paths['overlay_before']],child_manifest=after_meta) + require(binding['pass_all'],'Actual after array differs from recomputed receipt') + import torch + require(os.environ['CUDA_VISIBLE_DEVICES']=='' and not torch.cuda.is_initialized(),'CPU verifier initialized CUDA') + e._check_files(pins);c.load_checkpoint(checkpoint,value['checkpoint_manifest_sha256']) + for file,sha in (('conditioning_epoch.py',E_SHA),('conditioning_preflight_compare.py',Q_SHA),('engineering_compare.py',CMP_SHA)): + require(digest(LAB/file)==sha,'Verifier dependency changed') + require(hashes=={n:digest(directory/n) for n in FILES} and own==digest(__file__),'Source/update changed during validation') + receipt=dict(schema='conditioning-resident-cpu-validation-prototype-v1',complete=True,epoch_authority_sha256=authority_sha, + parent_checkpoint_manifest_sha256=value['checkpoint_manifest_sha256'],update_files_sha256=hashes, + before_overlay_sha256=pins[paths['overlay_before']],before_manifest_sha256=pins[paths['overlay_manifest_before']], + after_overlay_sha256=hashes['delta-after.npz'],after_manifest_sha256=hashes['overlay-manifest.json'], + before_delta_raw_sha256=r._raw(before['plastic_delta_mv']),after_delta_raw_sha256=r._raw(after['plastic_delta_mv']), + update_receipt_payload_sha256=update['payload_sha256'],validator_source=dict(path=str(Path(__file__).resolve()),sha256=own), + frozen_U_source_sha256=U_SHA,frozen_U_verify_receipt_recomputed=True,actual_sealed_events_used=True, + full_candidate_mapping_verified=True,brain_executed=False,learning_pass_claimed=False, + resident_prototype=True,original_validator_source=dict(path=str(ORIGINAL_VALIDATOR),sha256=ORIGINAL_VALIDATOR_SHA), + resident_cache_scope='SHA-verified imported modules only; no input/hash/state/delta/checkpoint caching') + e._save(out,receipt) + return receipt + + +def validate(epoch_authority,authority_sha,update_dir,chain_sha,receipt_file): + """Same arguments as frozen validate; exclusive receipt, no retry/reset API.""" + global _RESIDENT_STATE,_TORCH_MODULE,_SELF_SOURCE_SHA + require(_RESIDENT_STATE in ('new','ready'),'Resident worker active or failed; restart/retry is not authorized') + first=_RESIDENT_STATE=='new';_RESIDENT_STATE='active' + try: + require(sys.dont_write_bytecode and 'runtime_light' not in sys.modules and + 'runtime_continuation' not in sys.modules,'Dedicated CPU-only python -B resident worker required') + if first: + require('torch' not in sys.modules,'First request must not inherit Torch/runtime execution') + os.environ['CUDA_VISIBLE_DEVICES']='' + else: + require(sys.modules.get('torch') is _TORCH_MODULE and _TORCH_MODULE is not None and + os.environ.get('CUDA_VISIBLE_DEVICES')=='' and not _TORCH_MODULE.cuda.is_initialized(), + 'Resident Torch identity/CUDA isolation changed') + require(digest(unlinked(ORIGINAL_VALIDATOR))==ORIGINAL_VALIDATOR_SHA,'Frozen validator source changed') + source_sha=digest(unlinked(__file__)) + if _SELF_SOURCE_SHA is None:_SELF_SOURCE_SHA=source_sha + require(source_sha==_SELF_SOURCE_SHA,'Resident source changed between requests') + result=_validate_checked(epoch_authority,authority_sha,update_dir,chain_sha,receipt_file) + module=sys.modules.get('torch') + require(module is not None and (first or module is _TORCH_MODULE) and + os.environ.get('CUDA_VISIBLE_DEVICES')=='' and not module.cuda.is_initialized() and + 'runtime_light' not in sys.modules and 'runtime_continuation' not in sys.modules, + 'Resident request violated isolation') + for name,(path,sha,loaded) in _LOAD_CACHE.items(): + require(sys.modules.get(name) is loaded and Path(loaded.__file__).resolve()==path and + digest(unlinked(path))==sha,'Resident dependency changed during request') + require(digest(unlinked(ORIGINAL_VALIDATOR))==ORIGINAL_VALIDATOR_SHA and + digest(unlinked(__file__))==source_sha,'Validator source changed during request') + _TORCH_MODULE=module;_RESIDENT_STATE='ready' + return result + except BaseException: + _RESIDENT_STATE='failed' + raise + + +if __name__=='__main__': + parser=argparse.ArgumentParser(description=__doc__) + parser.add_argument('--authority',required=True);parser.add_argument('--authority-sha256',required=True) + parser.add_argument('--update-dir',required=True);parser.add_argument('--chain-sha256',required=True) + parser.add_argument('--receipt',required=True);args=parser.parse_args() + validate(args.authority,args.authority_sha256,args.update_dir,args.chain_sha256,args.receipt) + print(json.dumps(dict(complete=True,receipt=args.receipt,sha256=digest(args.receipt))),flush=True) + diff --git a/tools/conditioning-perf-lab-v1/test_perf_probe.py b/tools/conditioning-perf-lab-v1/test_perf_probe.py new file mode 100644 index 0000000..10a3481 --- /dev/null +++ b/tools/conditioning-perf-lab-v1/test_perf_probe.py @@ -0,0 +1,138 @@ +"""CPU-only synthetic safety/AST tests; not scientific performance evidence.""" +import ast +import copy +import hashlib +import importlib.util +import json +from pathlib import Path +import sys +import tempfile +import time +import unittest +from unittest.mock import patch + +import numpy as np +import perf_probe as P + + +def fresh_validator(): + path = Path(P.__file__).parent/'resident_validator.py' + spec = importlib.util.spec_from_file_location('_test_resident_fixture', path) + module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) + return module + + +class Tests(unittest.TestCase): + def test_safe_cache_reuses_same_verified_module(self): + v = fresh_validator() + with tempfile.TemporaryDirectory() as d: + p = Path(d)/'m.py'; p.write_text('VALUE = 7\n', encoding='utf-8'); sha = P.digest(p) + try: + a = v.load('_perf_unit_one', p, sha) + self.assertIs(a, v.load('_perf_unit_one', p, sha)) + p.write_text('VALUE = 8\n', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'bytes changed'): v.load('_perf_unit_one', p, sha) + finally: sys.modules.pop('_perf_unit_one', None) + + def test_module_identity_substitution_rejected(self): + v = fresh_validator() + with tempfile.TemporaryDirectory() as d: + p = Path(d)/'m.py'; p.write_text('VALUE=1', encoding='utf-8'); sha = P.digest(p) + try: + v.load('_perf_unit_two', p, sha); sys.modules['_perf_unit_two'] = object() + with self.assertRaisesRegex(ValueError, 'identity differs'): v.load('_perf_unit_two', p, sha) + finally: sys.modules.pop('_perf_unit_two', None) + + def test_inherited_torch_fails_and_poisoned_worker_stays_failed(self): + v = fresh_validator() + with patch.dict(sys.modules, {'torch': object()}): + with self.assertRaisesRegex(ValueError, 'must not inherit'): v.validate('x','x','x','x','x') + self.assertEqual(v._RESIDENT_STATE, 'failed') + with self.assertRaisesRegex(ValueError, 'active or failed'): v.validate('x','x','x','x','x') + + def test_ordinary_runtime_rejected(self): + v = fresh_validator() + with patch.dict(sys.modules, {'runtime_light': object()}): + with self.assertRaisesRegex(ValueError, 'Dedicated CPU-only'): v.validate('x','x','x','x','x') + self.assertEqual(v._RESIDENT_STATE, 'failed') + + def test_bitwise_signed_zero(self): + v = fresh_validator() + self.assertFalse(v.same(np.array([0.],np.float32), np.array([-0.],np.float32))) + self.assertFalse(v.same(np.array([0.],np.float32), np.array([0.],np.float64))) + + def test_copied_validation_body_AST_preserved(self): + original = P.VALIDATOR + if not original.exists(): + original = Path(P.__file__).parent.parent/'conditioning-execution-v1/validate_update_cpu.py' + old = ast.parse(original.read_text(encoding='utf-8')) + new = ast.parse((Path(P.__file__).parent/'resident_validator.py').read_text(encoding='utf-8')) + of = {n.name:n for n in old.body if isinstance(n,ast.FunctionDef)} + nf = {n.name:n for n in new.body if isinstance(n,ast.FunctionDef)} + for name in ('require','digest','same','unlinked','chain_binding'): + self.assertEqual(ast.dump(of[name]), ast.dump(nf[name])) + a,b=copy.deepcopy(of['validate']),copy.deepcopy(nf['_validate_checked']) + a.body=a.body[2:];b.name='validate' + for node in ast.walk(b): + if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id=='receipt' for t in node.targets): + node.value.keywords=[k for k in node.value.keywords if k.arg not in ('resident_prototype','original_validator_source','resident_cache_scope')] + next(k for k in node.value.keywords if k.arg=='schema').value=ast.Constant('conditioning-update-cpu-validation-v1') + self.assertEqual(ast.dump(a),ast.dump(b)) + + def fixture_receipts(self, root): + original={'path':str(P.VALIDATOR),'sha256':P.V_SHA} + candidate=Path(P.__file__).resolve().parent/'resident_validator.py' + a={'schema':'conditioning-update-cpu-validation-v1','validator_source':original, + 'complete':True,'test_scientific_value':{'raw_hash':'a'*64}} + b=copy.deepcopy(a);b.update(schema='conditioning-resident-cpu-validation-prototype-v1', + validator_source={'path':str(candidate),'sha256':P.digest(candidate)}, + original_validator_source=original,resident_prototype=True, + resident_cache_scope='SHA-verified imported modules only; no input/hash/state/delta/checkpoint caching') + p,q=Path(root)/'original.json',Path(root)/'prototype.json' + P.save(p,a);P.save(q,b) + return p,q,b + + def test_only_explicit_provenance_differences_allowed(self): + with tempfile.TemporaryDirectory() as d: + p,q,b=self.fixture_receipts(d) + self.assertTrue(P.compare_validation(p,q)['all_scientific_fields_identical']) + b['test_scientific_value']['raw_hash']='b'*64 + q.write_text(json.dumps(b),encoding='utf-8') + with self.assertRaisesRegex(ValueError,'Scientific verifier'): P.compare_validation(p,q) + + def test_provenance_substitution_rejected(self): + with tempfile.TemporaryDirectory() as d: + p,q,b=self.fixture_receipts(d);b['original_validator_source']['sha256']='0'*64 + q.write_text(json.dumps(b),encoding='utf-8') + with self.assertRaisesRegex(ValueError,'original-source'): P.compare_validation(p,q) + + def test_exclusive_save_no_overwrite(self): + with tempfile.TemporaryDirectory() as d: + p=Path(d)/'record.json';P.save(p,{'one':1}) + with self.assertRaises(FileExistsError): P.save(p,{'two':2}) + self.assertEqual(P.read(p),{'one':1}) + + def test_parent_traversal_rejected(self): + with self.assertRaisesRegex(ValueError, 'Parent traversal'): P.safe('/tmp/test/../escape') + + def test_absolute_deadline_enforced(self): + process = object.__new__(P.Process); process.deadline = time.perf_counter()-1 + with self.assertRaisesRegex(ValueError, 'absolute wall'): process.remaining() + + def test_bad_ready_child_is_reaped_before_constructor_raises(self): + children=[]; original=P.subprocess.Popen + def create_test_child(argv, **kwargs): + child=original([sys.executable,'-B','-c','import time; time.sleep(30)'],**kwargs) + children.append(child);return child + with tempfile.TemporaryDirectory() as d: + root=Path(d) + for name in ('logs','processes'): (root/name).mkdir() + plan=root/'plan.json';P.save(plan,{}) + with patch.object(P.subprocess,'Popen',side_effect=create_test_child), patch.object(P.Process,'line',side_effect=ValueError('bad ready fixture')): + with self.assertRaisesRegex(ValueError,'bad ready fixture'): + P.Process('update','server',plan,root,'bad-start',time.perf_counter()+10) + self.assertEqual(len(children),1);self.assertIsNotNone(children[0].poll()) + self.assertIsNotNone(P.read(root/'processes/bad-start.aborted.json')['actual_parent_observed_exit']) + + +if __name__=='__main__': unittest.main()