diff --git a/models/male-cns-lif/conditioning-interface-v2/PROTOCOL.md b/models/male-cns-lif/conditioning-interface-v2/PROTOCOL.md new file mode 100644 index 0000000..70ab6a6 --- /dev/null +++ b/models/male-cns-lif/conditioning-interface-v2/PROTOCOL.md @@ -0,0 +1,81 @@ +# Phase 4AP: exact, process-isolated conditioning interface + +This independent engineering branch repairs process ownership, not neural +mathematics. Phase 4AO remains sealed `FAIL-POSTPROCESS-MODULE-IDENTITY`. +It is not resumed or reclassified. No source in its 51-file inventory changes. + +## Registered scope + +The new registration covers 12 trial slots: **one explicitly imported, complete +AO neural checkpoint and 11 new neural executions**, 14,400 covered ticks and +13,200 new ticks. The imported slot is D1/reference/epoch-000. Its original case, +requests, spec and checkpoint paths remain unchanged. Incomplete AO process +evidence is never replaced by a successful AP process claim. + +Reuse both sealed AO request arrays byte-for-byte. Zero new sampler calls; +no seed selection, resampling, rate/parameter rescue or data-selected retry. +The two A epochs, original full 63-bit seeds, fixed PAM15 compartment, timing, +D1/D0/frozen arms and reference/light recording modes are inherited exactly. +New case specs differ only in output budget root/cap, with per-field derivations. +Each epoch starts a fresh neural state. Only its own arm/mode's preceding delta +is carried. This does not implement retention-state transition. + +## Process ownership + +Each new brain trial runs in a new `python -B` exec process using the frozen +ordinary runtime. It seals its checkpoint before returning. The separate CPU +update process loads only the frozen epoch adapter E and its verified dependencies, +with CUDA hidden before scientific imports. The coordinator never calls E. +No module binding is deleted, no private cache is cleared, no verifier is relaxed. + +The imported trial also receives fresh CPU postprocessing into AP, with no brain +rerun. Its delta raw float32 bits and update receipt must match the separately +sealed AO isolation diagnostic. NPZ container byte equality is not required. + +## Integrity and admission + +Commit all five new source files before preregistration. Bind actual input/source +and prepared-case SHA-256, prior seals, source commit, actual saved requests, +and the imported authority. Pin external registration SHA before executing. +Job SHA is externally passed to each child. Verify input/source bytes before +and after execution. Bind the complete output hash inventory, checkpoint/chain, +authority, parent delta, actual child exit status and logs. Exclusive attempt, +output and receipt paths prevent retry after partial failure. + +Single physical RTX 4090 UUID GPU-79f3a85d-adce-7833-7003-6488140ca596 only. +Check all three existing ComfyUI queues before every brain trial; never stop or +evict their services. Preserve precision configuration. Finite AP output cap +4 GiB, reserve 10 GiB, child timeout 1,800 seconds. Stop on first failure and +preserve all partial evidence. Never restart because the observing SSH command +or tool times out; inspect actual process/log state first. + +## Pass gate and interpretation + +- Six reference/light pairs: exact observed spikes, counters, final state, + delay state and delta; inherited Q gate with independent expected effective + weight hash computed from frozen base, frozen fast-APL mask and actual delta. +- Six zero-weight cross-arm comparisons: D0/frozen in each epoch and + D1/D0 before any update, separately for each recording mode. +- Original U receipt recomputation, correct compartment/mask/floor, exact + D0/frozen positive-zero delta and actual D1 nonzero update coverage. +- Second D1 epoch actually installs its own preceding delta file/raw bits. +- All previous sealed AO/diagnostic/AN/AM and 28 protected roots remain intact. + +If all checks hold, status is +`PASS-EXACT-PROCESS-ISOLATED-DELTA-INTERFACE-NOT-LEARNING`. +If D1 has no actual nonzero update coverage, report +`INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE`, not a synthetic success. +Failures remain failures; do not rerun with alternative inputs. + +Scheduled D is an artificial unconditioned-signal scalar, not a measured dopamine +concentration and not multiplied by accepted DAN events. Report requested, +accepted/rejected US events and actual DAN spikes separately. Unknown anatomical +mapping stays unknown. The frozen APL approximation and its non-KC limitations +remain disclosed. No intermediate light-mode state is claimed observed. + +This finite gate is **not learning success**, an A/B behavior evaluation, +retention, full acquisition or a learned-model candidate. Full conditioning +still needs its bounded registered gate/resource plan and validated post-update +same-state transition. Phase 4C's five trial evidence is preserved via AM seals; +this gate does not claim it reruns those trials. No game, RL, decoder, visual or +Web work is included. diff --git a/models/male-cns-lif/conditioning-interface-v2/coordinator.py b/models/male-cns-lif/conditioning-interface-v2/coordinator.py new file mode 100644 index 0000000..e138ddc --- /dev/null +++ b/models/male-cns-lif/conditioning-interface-v2/coordinator.py @@ -0,0 +1,559 @@ +"""AP: process-isolated delta interface, 1 imported + 11 new neural trials. + +Old AO remains sealed FAIL. This coordinator never calls E in its process and +does not mutate any frozen module globals. Neural and update workers use true +exec subprocesses. Source, plan and sampled input authority precede execution. +""" +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import time +import traceback + +import numpy as np + +MODEL = Path(__file__).resolve().parent +LAB = MODEL.parent / 'olfactory-conditioning-lab-v1' +sys.path.insert(0, str(LAB)) +import conditioning_preflight_prepare as OLD +import conditioning_preflight_workflow as AOW +import engineering_prepare as P +import engineering_workflow as ENV +import engineering_compare as COMP +import conditioning_preflight_compare as Q +import runtime_light as R +import runtime_continuation as C + +ROOT = Path('/data/digital-drosophila') +EXP = ROOT / 'experiments/phase4ap-conditioning-interface-v2' +AO = ROOT / 'experiments/phase4ao-conditioning-interface' +DIAG = ROOT / 'experiments/phase4ao-postprocess-isolation' +FIRST = 'D1--reference--epoch-000' +ANCHORS = { + str(AO / 'protocol/preregistered.json'): '9e2ed05879ce279da7dc0b4cac09a44d67665c230b56db94e2ca39c645bdc250', + str(AO / 'manifest.json'): 'dcfb24338a39ce0bb4d0bf5fe3c92a1907fb63536b4c57856c2f40203996f979', + str(AO / 'audit/final-status.json'): 'f8fff2cb7d4eb0abfc9653c989781c0310b6ad7ec42cce4949bae1a15fd793d8', + str(AO / 'runs' / FIRST / C.MANIFEST): 'e2ad8068bcc8450840ff4c5fb5fbe6a39a1960cd5869b5e16096fb579df28a3a', + str(AO / 'receipts' / (FIRST + '.authority.json')): '9c86c505d6156ca4ba0496105a7fde0556c0a05bfbd68dec6e7ec7c1865cdb60', + str(DIAG / 'manifest.json'): '202efff603a05016b3f530df9eb1d3c52e1c4119dbc4372ad7b75c6ca69013a2', + str(DIAG / 'diagnostic-receipt.json'): 'cefe84ac1642285b2c815a1f2a03677be064dd38125024f638eda4892cf477b3', +} +AO_REG = ANCHORS[str(AO / 'protocol/preregistered.json')] +BUDGET = dict(output_budget_bytes=4 * 1024**3, minimum_free_reserve_bytes=10 * 1024**3, + child_timeout_seconds=1800) +require, read, save, digest = P.require, P.read, P.save, P.digest + + +def same(a, b): return OLD.same(a, b) +def npz(p): return AOW.npz(p) + + +def check_pins(pins): + OLD.verify_pins(pins) + + +def helper(): + return P.load_module('phase4ap_seal_helper', ROOT / P.AI_HELPER_REL, P.AI_HELPER_SHA) + + +def previous_integrity(): + check_pins(ANCHORS) + h = helper() + result = dict(AO=h.verify_seal(AO), diagnostic=h.verify_seal(DIAG), AN=h.verify_seal(OLD.AN)) + old = P.read_registration(OLD.AN_REG) + result['AM'] = P.verify_authority(old['anchors']) + result['old_roots'] = P.verify_old_integrity(old) + require(read(AO / 'audit/final-status.json')['status'] == 'FAIL-POSTPROCESS-MODULE-IDENTITY', 'AO failure must remain unchanged') + return result + + +def derive_spec(original, budget_root, cap): + """The only new-case differences from AO are administrative budget fields.""" + require(type(budget_root) is str and Path(budget_root).is_absolute() and type(cap) is int and cap > 0, + 'Explicit new absolute budget/cap required') + require(budget_root != original['budget_root'], 'New experiment cannot reuse the stopped budget root') + spec = copy.deepcopy(original) + spec.update(budget_root=budget_root, output_budget_bytes=cap) + changes = {k: dict(before=original[k], after=spec[k]) for k in original if not same(original[k], spec[k])} + require(set(changes) <= {'budget_root', 'output_budget_bytes'}, 'Frozen neural/input spec changed') + return spec, dict(changes=changes, all_other_fields_exact=True, no_neural_parameter_change=True) + + +def source_inventory(): + names = ('coordinator.py', 'test_coordinator.py', 'isolated_workers.py', 'test_isolated_workers.py', 'PROTOCOL.md') + return {str(MODEL / n): digest(MODEL / n) for n in names} + + +def prepare(): + require(sys.dont_write_bytecode and EXP.is_dir() and not any(EXP.iterdir()), 'Use -B and the new empty AP directory') + before = previous_integrity() + old = OLD.read_registration(AO_REG) + generator = AOW.generated_inputs(AO_REG) + checkpoint = C.load_checkpoint(AO / 'runs' / FIRST, ANCHORS[str(AO / 'runs' / FIRST / C.MANIFEST)]) + require(checkpoint['fresh'] and checkpoint['steps'] == 1200, 'Imported trajectory is not complete/fresh') + source = source_inventory() + commit = subprocess.check_output(['git', '-C', str(ROOT), 'rev-parse', 'HEAD']).decode().strip() + require(len(commit) == 40, 'Git code commit missing') + for path, sha in source.items(): + blob = subprocess.check_output(['git', '-C', str(ROOT), 'show', commit + ':' + Path(path).relative_to(ROOT).as_posix()]) + require(hashlib.sha256(blob).hexdigest() == sha, 'AP sources must be committed before registration') + pins = dict(ANCHORS) + for field in ('source_sha256', 'input_sha256', 'prepared_file_sha256'): + pins.update(old[field]) + pins.update(generator['inputs_sha256']) + pins[str(AO / 'receipts/input-generation.json')] = digest(AO / 'receipts/input-generation.json') + pins[str(AO / 'receipts/input-generation.process.json')] = digest(AO / 'receipts/input-generation.process.json') + for p in (DIAG / 'postprocess').iterdir(): + pins[str(p)] = digest(p) + check_pins(pins) + for n in ('protocol/source-snapshot', 'cases', 'runs', 'updates', 'jobs', 'receipts', 'logs', 'audit', 'regression'): + (EXP / n).mkdir(parents=True) + for path in source: + shutil.copyfile(path, EXP / 'protocol/source-snapshot' / Path(path).name) + save(EXP / 'regression/integrity-before.json', before) + slots = [] + for row in old['plan']['calls']: + row = copy.deepcopy(row) + imported = row['row_id'] == FIRST + original_case = Path(old['cases'][row['row_id']]) + if imported: + case, run = original_case, AO / 'runs' / FIRST + else: + case, run = EXP / 'cases' / row['row_id'], EXP / 'runs' / row['row_id'] + case.mkdir() + spec, change = derive_spec(read(original_case / 'spec.json'), str(EXP), BUDGET['output_budget_bytes']) + R.validate_spec(spec, case) + save(case / 'spec.json', spec) + save(case / 'derivation.json', dict(original_spec=str(original_case / 'spec.json'), + original_spec_sha256=digest(original_case / 'spec.json'), **change)) + with (original_case / 'requests.npz').open('rb') as src, (case / 'requests.npz').open('xb') as dst: + shutil.copyfileobj(src, dst) + require(digest(case/'requests.npz') == digest(original_case/'requests.npz'), 'Saved requests copy differs') + row.update(origin='imported_sealed_AO_neural_trajectory' if imported else 'new_AP_neural_execution', + case_dir=str(case), neural_dir=str(run), original_case_dir=str(original_case)) + slots.append(row) + prepared = {str(p): digest(p) for sub in ('protocol/source-snapshot', 'cases', 'regression') + for p in (EXP / sub).rglob('*') if p.is_file()} + reg = dict(schema='phase4ap-process-isolated-interface-v1', anchors=ANCHORS, source_sha256=source, + input_sha256=pins, prepared_file_sha256=prepared, source_code_commit=commit, budget=BUDGET, + slots=slots, selected_epochs=old['plan']['selected_epochs'], compartment=old['plan']['compartment_verbatim'], + dan_type=old['plan']['dan_type'], initial_overlay=old['initial_overlay'], authority_template=str(AO / 'receipts' / (FIRST + '.authority.json')), + imported_slot=FIRST, imported_checkpoint_sha256=ANCHORS[str(AO / 'runs' / FIRST / C.MANIFEST)], + covered_slots=12, new_neural_calls=11, covered_ticks=14400, new_neural_ticks=13200, new_sampler_calls=0, + original_failed_workflow_reclassified=False, inherited_neural_checkpoint_complete=True, + original_AO_process_complete=False, first_postprocess='fresh exec CPU E-only process, compare to sealed independent diagnostic', + neural_reset='fresh each epoch; carry delta only', update_trace_backend='torch', within_trial_updates=False, + scheduled_D='artificial scalar not multiplied by actual DAN acceptance', + success='six exact reference/light pairs, six exact zero-weight cross-arm pairs, all real v2 updates recomputed, D0/frozen exact positive-zero, nonzero D1 coverage and epoch 2 exact installation', + no_nonzero_coverage_status='INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE', + failure_policy='stop and preserve partials; no automatic child retries, seed search or parameter rescue', + full_conditioning=False, retention_transition_supported=False, learning_success_claimed=False, + light_intermediate_states_observed=False, future_hashes_fabricated=False) + check_pins({**pins, **source, **prepared}) + save(EXP / 'protocol/preregistered.json', reg) + print('REGISTRATION_SHA256', digest(EXP / 'protocol/preregistered.json'), flush=True) + + +def registration(sha): + require(P.sha(sha) and digest(EXP / 'protocol/preregistered.json') == sha, 'Registration differs from external SHA') + reg = read(EXP / 'protocol/preregistered.json') + require(reg['schema'] == 'phase4ap-process-isolated-interface-v1' and reg['anchors'] == ANCHORS and reg['budget'] == BUDGET, + 'Wrong AP registration') + require(len(reg['slots']) == 12 and sum(r['origin'] == 'new_AP_neural_execution' for r in reg['slots']) == 11, + 'AP must cover exactly one import plus eleven new neural calls') + check_pins({**reg['input_sha256'], **reg['source_sha256'], **reg['prepared_file_sha256']}) + return reg + + +def slot(reg, name): + matches = [r for r in reg['slots'] if r['row_id'] == name] + require(len(matches) == 1, 'Unknown/nonunique slot') + return matches[0] + + +def paths(row): + name = row['row_id'] + require(type(name) is str and name and all(c.isalnum() or c in '-_' for c in name), 'Unsafe slot identity') + return dict(neural=Path(row['neural_dir']), case=Path(row['case_dir']), update=EXP / 'updates' / name, + authority=EXP / 'jobs' / (name + '.authority.json'), record=EXP / 'receipts' / (name + '.slot.json')) + + +def admission(reg): + require(not any((EXP / x).exists() for x in ('audit/stopped.json', 'audit/execution-complete.json', 'manifest.json')), + 'Stopped/completed AP must not be resumed automatically') + all_paths = list(EXP.rglob('*')) + require(not any(p.is_symlink() for p in all_paths), 'Linked output is forbidden') + used = sum(p.stat().st_size for p in all_paths if p.is_file()) + require(used <= reg['budget']['output_budget_bytes'] and shutil.disk_usage(EXP).free >= + reg['budget']['minimum_free_reserve_bytes'] + max(0, reg['budget']['output_budget_bytes'] - used), 'AP resource cap exceeded') + + +def before_overlay(reg, reg_sha, row): + previous = row['previous_delta_row_id'] + if previous is None: return reg['initial_overlay'] + parent = slot(reg, previous) + require(parent['arm'] == row['arm'] and parent['diagnostic_mode'] == row['diagnostic_mode'] and + parent['input_epoch_index'] + 1 == row['input_epoch_index'], 'Invalid delta ancestry') + verify_slot(reg, reg_sha, parent) + directory = paths(parent)['update'] + return dict(overlay_file=str(directory / 'delta-after.npz'), overlay_manifest=str(directory / 'overlay-manifest.json')) + + +def authority_for(reg, row, checkpoint_sha, overlay): + value = copy.deepcopy(read(reg['authority_template'])) + p = paths(row) + value.update(epoch_id=row['epoch_id'], diagnostic_mode=row['diagnostic_mode'], checkpoint_dir=str(p['neural']), + checkpoint_manifest_sha256=checkpoint_sha) + value['command'] = dict(steps=row['steps'], compartment=reg['compartment'], + us_tick=reg['selected_epochs'][row['input_epoch_index']]['us']['tick'], d_command=row['d_command'], + frozen=row['frozen'], trace_backend=reg['update_trace_backend']) + changed = dict(spec=p['case'] / 'spec.json', requests=p['case'] / 'requests.npz', state=p['neural'] / 'state.npz', + overlay_before=overlay['overlay_file'], overlay_manifest_before=overlay['overlay_manifest']) + for role, path in changed.items(): + value['files'][role] = dict(path=str(path), sha256=digest(path)) + return value + + +def job_for(reg, reg_sha, row, kind, overlay=None): + require(kind in ('brain', 'update'), 'Unknown child kind') + p = paths(row); name = row['row_id'] + '.' + kind + worker = MODEL / 'isolated_workers.py' + pins = {**reg['input_sha256'], **reg['source_sha256'], **reg['prepared_file_sha256']} + base = dict(schema='phase4ap-isolated-worker-job-v1', job_id=name, kind=kind, + registration_sha256=reg_sha, worker_sha256=reg['source_sha256'][str(worker)], + output_dir=str(p['neural'] if kind == 'brain' else p['update']), + receipt_file=str(EXP / 'receipts' / (name + '.worker.json'))) + if kind == 'brain': + require(row['origin'] == 'new_AP_neural_execution', 'Imported neural trajectory must not be executed again') + for path in (p['case'] / 'spec.json', p['case'] / 'requests.npz', overlay['overlay_file'], overlay['overlay_manifest']): + pins[str(path)] = digest(path) + base.update(case_dir=str(p['case']), build_root=str(EXP / 'extension-build'), **overlay, + diagnostic_mode=row['diagnostic_mode'], admission=dict(gpu_uuid=ENV.GPU, + comfy_queue_urls=[f'http://192.168.10.51:{port}/queue' for port in (8188, 8189, 8190)]), + sources={role: dict(path=str(LAB / (role + '.py')), sha256=reg['input_sha256'][str(LAB / (role + '.py'))]) + for role in ('runtime_light', 'runtime_continuation')}) + else: + authority = read(p['authority']) + pins[str(p['authority'])] = digest(p['authority']) + pins.update({v['path']: v['sha256'] for v in authority['files'].values()}) + pins.update(authority['frozen_inputs']) + pins[str(p['neural'] / C.MANIFEST)] = authority['checkpoint_manifest_sha256'] + base.update(authority_file=str(p['authority']), authority_sha256=digest(p['authority']), + sources={'conditioning_epoch': dict(path=str(LAB / 'conditioning_epoch.py'), + sha256=reg['input_sha256'][str(LAB / 'conditioning_epoch.py')])}) + base['input_sha256'] = pins + return base + + +def process_locations(row, kind): + name = row['row_id'] + '.' + kind + return dict(job=EXP / 'jobs' / (name + '.json'), worker=EXP / 'receipts' / (name + '.worker.json'), + process=EXP / 'receipts' / (name + '.process.json'), stdout=EXP / 'logs' / (name + '.stdout.log'), + stderr=EXP / 'logs' / (name + '.stderr.log')) + + +def launch(reg, reg_sha, row, kind, job): + loc = process_locations(row, kind) + require(not any(p.exists() for p in loc.values()), 'Partial worker cannot be retried') + admission(reg) + save(loc['job'], job) + command = [sys.executable, '-B', str(MODEL / 'isolated_workers.py'), kind, '--job', str(loc['job']), + '--job-sha256', digest(loc['job'])] + start = time.perf_counter() + env = dict(os.environ) + if kind == 'update': env['CUDA_VISIBLE_DEVICES'] = '' + timed_out = False + with loc['stdout'].open('xb') as out, loc['stderr'].open('xb') as err: + try: + result = subprocess.run(command, cwd=MODEL, env=env, stdout=out, stderr=err, + timeout=reg['budget']['child_timeout_seconds']) + exit_code = result.returncode + except subprocess.TimeoutExpired: + timed_out, exit_code = True, None + complete = not timed_out and exit_code == 0 and loc['worker'].is_file() + receipt = dict(complete=complete, registration_sha256=reg_sha, row_id=row['row_id'], kind=kind, + exit_code=exit_code, timed_out=timed_out, argv=command, job_sha256=digest(loc['job']), + worker_receipt_sha256=digest(loc['worker']) if loc['worker'].is_file() else None, stdout_sha256=digest(loc['stdout']), + stderr_sha256=digest(loc['stderr']), wall_seconds=time.perf_counter()-start) + save(loc['process'], receipt) + require(complete, 'Isolated child failed or timed out; preserve partials') + return verify_process(reg, reg_sha, row, kind, job) + + +def verify_process(reg, reg_sha, row, kind, expected_job): + loc = process_locations(row, kind) + process = read(loc['process']) + require(process['complete'] is True and process['exit_code'] == 0 and process['registration_sha256'] == reg_sha and + process['row_id'] == row['row_id'] and process['kind'] == kind, 'Missing actual complete child process') + for k, f in (('job_sha256','job'), ('worker_receipt_sha256','worker'), ('stdout_sha256','stdout'), ('stderr_sha256','stderr')): + require(process[k] == digest(loc[f]), 'Child evidence changed: ' + k) + require(same(read(loc['job']), expected_job), 'Job differs from registered slot and actual parent') + worker = read(loc['worker']) + require(worker['schema'] == 'phase4ap-isolated-worker-receipt-v1' and worker['complete'] is True and + worker['job_sha256'] == process['job_sha256'] and worker['registration_sha256'] == reg_sha and + worker['job_id'] == expected_job['job_id'] and worker['kind'] == kind and + worker['worker_sha256'] == expected_job['worker_sha256'] and + worker['output_dir'] == expected_job['output_dir'] and worker['receipt_file'] == str(loc['worker']) and + worker['source_files'] == expected_job['sources'], 'Worker is not bound to this job') + attempt = Path(str(loc['worker']) + '.attempt.json') + require(digest(attempt) == worker['attempt_sha256'], 'Worker attempt record changed') + expected_pins = {**expected_job['input_sha256'], str(loc['job']): process['job_sha256'], + str(MODEL/'isolated_workers.py'): expected_job['worker_sha256']} + require(worker['verified_input_sha256'] == expected_pins, 'Worker did not verify the complete job inputs') + output = Path(expected_job['output_dir']) + descendants = list(output.rglob('*')) + require(not any(p.is_symlink() for p in descendants), 'Linked worker output forbidden') + actual_outputs = {p.relative_to(output).as_posix(): digest(p) for p in descendants if p.is_file()} + require(actual_outputs and actual_outputs == worker['output_files_sha256'], 'Worker output inventory changed') + if kind == 'brain': + require(worker['result']['complete_fresh_trial'] is True and + worker['result']['steps'] == row['steps'] and + worker['result']['checkpoint_manifest_sha256'] == digest(output/C.MANIFEST), 'Brain worker checkpoint mismatch') + ENV.verify_environment(worker['environment'], read(output/'summary.json')['apl_component']['environment']) + else: + result = worker['result']; chain = read(output/'chain-receipt.json') + require(result['brain_executed'] is False and result['weights_installed'] is False and + result['chain_receipt_sha256'] == digest(output/'chain-receipt.json') and + chain['authority_sha256'] == expected_job['authority_sha256'], 'Update worker chain mismatch') + for field in ('source_neural_checkpoint_manifest_sha256', 'before_overlay_file_sha256', + 'after_overlay_file_sha256', 'before_overlay_manifest_sha256', 'after_overlay_manifest_sha256'): + require(result[field] == chain[field], 'Update worker result differs: ' + field) + require(chain['after_overlay_file_sha256'] == digest(output/'delta-after.npz') and + chain['after_overlay_manifest_sha256'] == digest(output/'overlay-manifest.json'), 'Update after-overlay payload differs') + require(worker['environment']['CUDA_hidden'] is True and worker['environment']['CUDA_initialized'] is False, + 'Update worker was not CPU-isolated') + check_pins(expected_job['input_sha256']) + return worker + + +def verify_slot(reg, reg_sha, row): + p = paths(row); record = read(p['record']) + require(record['complete'] is True and record['registration_sha256'] == reg_sha and same(record['row'], row), 'Slot record mismatch') + for name, sha in record['files_sha256'].items(): require(digest(name) == sha, 'Slot output changed') + checkpoint = C.load_checkpoint(p['neural'], record['checkpoint_manifest_sha256']) + require(checkpoint['fresh'] and checkpoint['steps'] == row['steps'], 'Slot is not a fresh neural epoch') + if row['origin'] == 'imported_sealed_AO_neural_trajectory': + require(record['checkpoint_manifest_sha256'] == reg['imported_checkpoint_sha256'] and + record['new_neural_execution'] is False and row['row_id'] == FIRST, 'Imported provenance mismatch') + else: + require(record['new_neural_execution'] is True, 'New runtime marked as import') + overlay = before_overlay(reg, reg_sha, row) + require(record['before_overlay_sha256'] == digest(overlay['overlay_file']), 'Slot parent delta changed') + if row['row_id'] != FIRST: + worker = verify_process(reg, reg_sha, row, 'brain', job_for(reg, reg_sha, row, 'brain', overlay)) + require(worker['result']['checkpoint_manifest_sha256'] == record['checkpoint_manifest_sha256'], 'Slot checkpoint differs from actual brain') + authority = authority_for(reg, row, record['checkpoint_manifest_sha256'], overlay) + require(same(authority, read(p['authority'])), 'Slot authority differs from registered trial') + verify_process(reg, reg_sha, row, 'update', job_for(reg, reg_sha, row, 'update')) + chain = read(p['update']/'chain-receipt.json') + require(chain['authority_sha256'] == digest(p['authority']) and + chain['source_neural_checkpoint_manifest_sha256'] == record['checkpoint_manifest_sha256'] and + chain['before_overlay_file_sha256'] == digest(overlay['overlay_file']), 'Slot update ancestry differs') + verify_delta_payload(overlay, p['update'], read(p['update']/'update-receipt.json')) + return record + + +def verify_delta_payload(overlay, directory, update): + _, after, _, _ = R.load_overlay(directory/'delta-after.npz', directory/'overlay-manifest.json') + binding = Q.audit_delta_binding(npz(overlay['overlay_file'])['plastic_delta_mv'], after['plastic_delta_mv'], + after['plastic_delta_mv'], update_receipt=update, parent_snapshot_sha256=digest(overlay['overlay_file']), + child_manifest=read(directory/'overlay-manifest.json')) + require(binding['pass_all'], 'Actual delta binding failed') + return binding, after['plastic_delta_mv'] + + +def verify_imported_delta(delta, update): + prior = npz(DIAG/'postprocess/delta-after.npz')['plastic_delta_mv'] + require(COMP.bit_equal(prior, delta) and same(update, read(DIAG/'postprocess/update-receipt.json')), + 'Imported trajectory postprocessing differs from sealed diagnostic') + + +def perform_slot(reg, reg_sha, row): + p = paths(row) + require(not p['record'].exists() and not p['authority'].exists() and not p['update'].exists(), 'Partial slot cannot be retried') + overlay = before_overlay(reg, reg_sha, row) + started = time.perf_counter() + if row['origin'] == 'imported_sealed_AO_neural_trajectory': + checkpoint_sha = reg['imported_checkpoint_sha256'] + C.load_checkpoint(p['neural'], checkpoint_sha) + else: + worker = launch(reg, reg_sha, row, 'brain', job_for(reg, reg_sha, row, 'brain', overlay)) + checkpoint_sha = worker['result']['checkpoint_manifest_sha256'] + C.load_checkpoint(p['neural'], checkpoint_sha) + ENV.verify_environment(worker['environment'], read(p['neural'] / 'summary.json')['apl_component']['environment']) + authority = authority_for(reg, row, checkpoint_sha, overlay) + save(p['authority'], authority) + launch(reg, reg_sha, row, 'update', job_for(reg, reg_sha, row, 'update')) + update = read(p['update'] / 'update-receipt.json') + AOW.replay_update(authority, update) # CPU-only original U verification, never E in this process. + binding, delta = verify_delta_payload(overlay, p['update'], update) + if row['row_id'] == FIRST: + verify_imported_delta(delta, update) + save(EXP / 'receipts' / (row['row_id'] + '.binding.json'), binding) + files = [p['authority'], *p['update'].iterdir(), EXP / 'receipts' / (row['row_id'] + '.binding.json')] + for kind in (('update',) if row['row_id'] == FIRST else ('brain', 'update')): + files.extend(process_locations(row, kind).values()) + files.append(Path(str(process_locations(row, kind)['worker']) + '.attempt.json')) + registration(reg_sha) + save(p['record'], dict(complete=True, registration_sha256=reg_sha, row=row, + new_neural_execution=row['row_id'] != FIRST, original_AO_workflow_pass_claimed=False, + checkpoint_manifest_sha256=checkpoint_sha, before_overlay_sha256=digest(overlay['overlay_file']), + files_sha256={str(f): digest(f) for f in files}, wall_seconds=time.perf_counter()-started, + update_recomputed=True, learning_success_claimed=False)) + verify_slot(reg, reg_sha, row) + + +def run(reg_sha): + reg = registration(reg_sha); admission(reg) + completed = [] + try: + for row in reg['slots']: + admission(reg) + print('START', row['row_id'], row['origin'], flush=True) + perform_slot(reg, reg_sha, row) + completed.append(row['row_id']) + print('COMPLETE', row['row_id'], flush=True) + save(EXP / 'audit/execution-complete.json', dict(registration_sha256=reg_sha, complete=True, + completed=completed, covered_slots=12, new_brain_calls=11, new_ticks=13200, covered_ticks=14400, + imported_trajectory_count=1, new_sampler_calls=0, learning_success_claimed=False)) + except Exception as error: + save(EXP / 'audit/stopped.json', dict(complete=False, registration_sha256=reg_sha, + error_type=type(error).__name__, error=str(error), traceback=traceback.format_exc(), + completed_slots=completed, no_automatic_retry=True, partials_preserved=True)) + raise + + +def expected_weight(reg, reg_sha, row): + overlay = before_overlay(reg, reg_sha, row) + h, plan, _, _ = R.load_overlay(overlay['overlay_file'], overlay['overlay_manifest']) + spec = read(paths(row)['case'] / 'spec.json'); arrays = {} + try: + for n in ('nodes', 'weights', 'edge_id', 'src'): + arrays[n] = np.load(Path(spec['graph_dir']) / (n+'.npy'), mmap_mode='r', allow_pickle=False) + require(np.array_equal(arrays['nodes'], np.arange(166700, dtype=np.int32)), 'Not canonical fullgraph nodes') + mask = np.isin(arrays['src'], np.load(Path(spec['apl_bundle_dir']) / 'apl_globals.npy', allow_pickle=False)) + require(int(mask.sum()) == 5845, 'Fast APL mask changed') + mapping = h.map_installation(arrays['weights'], arrays['edge_id'], mask, plan) + require(mapping['missing_candidate_count'] == 0, 'Incomplete mask mapping') + return h.expected_hashes(arrays['weights'], mask, mapping)[1] + finally: + for a in arrays.values(): a._mmap.close() + + +def analyze(reg_sha): + reg = registration(reg_sha) + complete = read(EXP / 'audit/execution-complete.json') + require(not (EXP/'audit/stopped.json').exists() and complete['complete'] and complete['registration_sha256'] == reg_sha and + complete['completed'] == [r['row_id'] for r in reg['slots']], 'Complete AP coverage is missing') + records, deltas, summaries = {}, {}, [] + for row in reg['slots']: + p=paths(row); records[row['row_id']] = verify_slot(reg, reg_sha, row) + overlay = before_overlay(reg, reg_sha, row) + if row['row_id'] != FIRST: + worker = verify_process(reg, reg_sha, row, 'brain', job_for(reg, reg_sha, row, 'brain', overlay)) + require(worker['result']['checkpoint_manifest_sha256'] == records[row['row_id']]['checkpoint_manifest_sha256'], 'Brain worker checkpoint mismatch') + authority = authority_for(reg, row, records[row['row_id']]['checkpoint_manifest_sha256'], overlay) + require(same(authority, read(p['authority'])), 'Actual command/input authority differs from registered row') + verify_process(reg, reg_sha, row, 'update', job_for(reg, reg_sha, row, 'update')) + update = read(p['update']/'update-receipt.json'); AOW.replay_update(authority, update) + chain = read(p['update']/'chain-receipt.json') + require(chain['authority_sha256'] == digest(p['authority']) and chain['source_neural_checkpoint_manifest_sha256'] == + records[row['row_id']]['checkpoint_manifest_sha256'], 'Update chain differs from actual checkpoint') + _, delta = verify_delta_payload(overlay, p['update'], update) + if row['row_id'] == FIRST: verify_imported_delta(delta, update) + deltas[row['row_id']]=delta + if row['arm'] in ('D0','frozen'): require(not np.any(delta.view(np.uint32)), 'Zero update controls are not exact positive-zero') + loaded=COMP.load_run(p['neural']) + try: + us=npz(AO/f"inputs/generated/epoch-{row['input_epoch_index']:03d}/us_requests.npz") + keys=us['tick'].astype(np.int64)*166700+us['neuron_idx'] + actual=loaded['requests']; selected=actual[np.isin(actual[:,0]*166700+actual[:,1], keys)] + require(len(selected)==len(keys), 'US delivery accounting incomplete') + dan=npz(AO/'inputs/static/populations.npz')[reg['dan_type']] + s=loaded['summary'] + summaries.append(dict(row_id=row['row_id'], origin=row['origin'], spikes=s['total_spikes'], + active_neurons=int(np.count_nonzero(loaded['state']['counts'])), all_finite=s['all_finite'], + peak_vram=s['peak_allocated_vram_bytes'], simulation_seconds=s['diagnostic_simulation_seconds'], + runtime_wall_seconds=s['wall_seconds'], slot_wall_seconds=records[row['row_id']]['wall_seconds'], + updates=update['counts'], DAN=dict(requested=len(selected),accepted=int(selected[:,3].sum()), + rejected=int(selected[:,4].sum()), actual_spikes=int(loaded['state']['counts'][dan].sum()), + scheduled_D=row['d_command'], actual_DAN_gates_update=False))) + finally: COMP.close_run(loaded) + pairs=[]; cross=[] + for arm in ('D1','D0','frozen'): + for epoch in range(2): + names=[f'{arm}--{mode}--epoch-{epoch:03d}' for mode in ('reference','light')] + runs=[COMP.load_run(paths(slot(reg,n))['neural']) for n in names] + try: + requests=npz(AO/f'inputs/generated/epoch-{epoch:03d}/requests.npz') + expected=expected_weight(reg,reg_sha,slot(reg,names[0])) + report=Q.compare_epoch_pair(*runs,expected_requests=np.column_stack((requests['tick'],requests['neuron_idx'])), + expected_steps=1200,expected_effective_weight_sha256=expected) + require(report['pass_all'] and COMP.bit_equal(deltas[names[0]],deltas[names[1]]), 'Reference/light regression failed') + pairs.append(dict(arm=arm,epoch=epoch,exact_delta_after=True,expected_effective_weight_sha256=expected,comparison=report)) + finally: + for r in runs:COMP.close_run(r) + for mode in ('reference','light'): + for left,right,epoch in (('D0','frozen',0),('D0','frozen',1),('D1','D0',0)): + runs=[COMP.load_run(paths(slot(reg,f'{a}--{mode}--epoch-{epoch:03d}'))['neural']) for a in (left,right)] + try: + result=COMP.compare_fresh(*runs); require(result['numerical_exact'],'Zero-weight cross-arm behavior differs') + cross.append(dict(left=left,right=right,epoch=epoch,mode=mode,comparison=result)) + finally: + for r in runs:COMP.close_run(r) + installations={} + for mode in ('reference','light'): + first=f'D1--{mode}--epoch-000'; second=f'D1--{mode}--epoch-001' + second_row=slot(reg,second); before=before_overlay(reg,reg_sha,second_row) + audit=read(paths(second_row)['neural']/'overlay-audit.json') + require(audit['delta_raw_sha256']==R._raw(deltas[first]) and + records[second]['before_overlay_sha256']==digest(before['overlay_file']), 'Second epoch did not install first delta exactly') + installations[mode]=dict(exact=True,nonzero_edges=int(np.count_nonzero(deltas[first]))) + coverage=all(s['updates']['actual_bitwise_changed_edges']>0 for s in summaries if s['row_id'].startswith('D1--')) + return dict(status='PASS-EXACT-PROCESS-ISOLATED-DELTA-INTERFACE-NOT-LEARNING' if coverage else 'INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE', + registration_sha256=reg_sha, covered_slots=12,new_neural_calls=11,imported_neural_trajectories=1,new_sampler_calls=0, + new_ticks=13200,covered_ticks=14400,pairs=pairs,cross_arm=cross,rows=summaries,epoch2_installations=installations, + D0_frozen_exact_zero=True,nonzero_D1_coverage=coverage,original_AO_status='FAIL-POSTPROCESS-MODULE-IDENTITY', + retention_transition_supported=False,full_conditioning_executed=False,learning_success_claimed=False,learned_candidate=False, + scope='One replicate, two A epochs, one PAM15 compartment; exact interface coverage only. No A/B behavioral evaluation or learning success.', + light_intermediate_states_observed=False) + + +def finalize(reg_sha): + require(not (EXP/'manifest.json').exists() and not (EXP/'audit/final-status.json').exists(), 'Never overwrite a final or partial seal') + report=analyze(reg_sha) + save(EXP/'audit/interface-analysis.json',report) + after=previous_integrity();save(EXP/'regression/integrity-after.json',after) + registration(reg_sha) + final=dict(status=report['status'],registration_sha256=reg_sha,analysis_sha256=digest(EXP/'audit/interface-analysis.json'), + covered_slots=12,new_neural_calls=11,imported_neural_trajectories=1,new_sampler_calls=0, + old_integrity_pass=True,old_integrity_roots=len(after['old_roots']),AO_failure_preserved=True, + source_integrity_pass=True,old_five_reference_trials='Sealed AM evidence retained; not rerun by this interface gate', + learned_candidate=False,learning_success_claimed=False,full_conditioning_executed=False) + save(EXP/'audit/final-status.json',final) + all_paths=sorted(EXP.rglob('*'));require(not any(p.is_symlink() or '__pycache__' in p.parts for p in all_paths),'Unsafe seal payload') + files=[dict(path=p.relative_to(EXP).as_posix(),bytes=p.stat().st_size,sha256=digest(p)) for p in all_paths if p.is_file()] + save(EXP/'manifest.json',dict(root=str(EXP),files=files)) + with (EXP/'checksums.sha256').open('x',encoding='utf-8',newline='\n') as out: + out.writelines(f['sha256']+' '+f['path']+'\n' for f in files) + out.write(digest(EXP/'manifest.json')+' manifest.json\n') + helper().verify_seal(EXP) + print(json.dumps(dict(status=final['status'],manifest_sha256=digest(EXP/'manifest.json'), + final_sha256=digest(EXP/'audit/final-status.json'))),flush=True) + + +if __name__=='__main__': + require(sys.dont_write_bytecode,'Use scientific Python -B') + parser=argparse.ArgumentParser();parser.add_argument('action',choices=('prepare','run','finalize')) + parser.add_argument('--registration-sha256');args=parser.parse_args() + if args.action=='prepare':prepare() + elif args.action=='run':run(args.registration_sha256) + else:finalize(args.registration_sha256) diff --git a/models/male-cns-lif/conditioning-interface-v2/isolated_workers.py b/models/male-cns-lif/conditioning-interface-v2/isolated_workers.py new file mode 100644 index 0000000..7041c3f --- /dev/null +++ b/models/male-cns-lif/conditioning-interface-v2/isolated_workers.py @@ -0,0 +1,381 @@ +"""Isolated AP workers; invoke each job in a NEW ``python -B`` process. + +Top-level imports are stdlib only. Brain loads the frozen ordinary runtime, +then its read-only checkpoint sealer; update loads ONLY the frozen epoch +adapter. Never import the old coordinator, share/patch loader caches, or use a +forked interpreter. The external job SHA binds the caller's authority; this +worker does not independently prove preregistration or anatomical provenance. + +Success receipts are outside all inputs/checkpoints. A separate exclusive +``.attempt.json`` prevents retries even after a pre-output failure. +The parent must additionally bind the real child exit code and stdout/stderr; +the receipt cannot certify a process exit that has not yet happened. +""" +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import time +import urllib.request + + +SCHEMA = 'phase4ap-isolated-worker-job-v1' +GPU = 'GPU-79f3a85d-adce-7833-7003-6488140ca596' +SCIENCE_BIN = '/data/digital-drosophila/models/male-cns-lif/.venv/bin' +COMFY_URLS = tuple(f'http://192.168.10.51:{port}/queue' for port in (8188, 8189, 8190)) +SOURCE_PINS = { + 'runtime_light': '522b8c31ea938890fe15364fccd25ccc5bded54101754909f255532e7ad01aae', + 'runtime_continuation': '0b8e8e0a6a23a079dc62a88355ce355d53b3ae22f6665519a8f6aef2dcfc5aba', + 'conditioning_epoch': '757935a4815d9f625640df685184ed3fe44a8845323562796da7aa43cbb194e1', +} +GRAPH_FILES = ('nodes.npy', 'edge_id.npy', 'src.npy', 'dst.npy', 'signed_raw.npy', 'ptr.npy', 'weights.npy') +APL_FILES = ('input_matrix.npy', 'output_matrix.npy', 'kc_globals.npy', 'apl_globals.npy') +COMMON = {'schema', 'job_id', 'kind', 'registration_sha256', 'worker_sha256', + 'output_dir', 'receipt_file', 'sources', 'input_sha256'} +BRAIN_FIELDS = {'case_dir', 'build_root', 'overlay_file', 'overlay_manifest', 'diagnostic_mode', 'admission'} +UPDATE_FIELDS = {'authority_file', 'authority_sha256'} + + +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 valid_sha(value): + return type(value) is str and re.fullmatch('[0-9a-f]{64}', value) is not None + + +def read_json(path): + def pairs(items): + result = {} + for key, value in items: + require(key not in result, 'Duplicate JSON key: ' + key) + result[key] = value + return result + return json.loads(Path(path).read_text(encoding='utf-8'), object_pairs_hook=pairs, + parse_constant=lambda x: (_ for _ in ()).throw(ValueError('Nonfinite JSON: ' + x))) + + +def save_json(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 absolute_path(value, *, kind=None): + require(type(value) is str and Path(value).is_absolute(), 'Explicit absolute path required') + path = Path(value) + for part in (path, *path.parents): + require(not part.is_symlink() and not (hasattr(part, 'is_junction') and part.is_junction()), + 'Linked/junction path is forbidden: ' + str(part)) + path = path.resolve(strict=kind is not None) + if kind == 'file': + require(path.is_file(), 'Regular input file required: ' + str(path)) + elif kind == 'dir': + require(path.is_dir(), 'Existing directory required: ' + str(path)) + return path + + +def verify_pins(pins): + for path, expected in pins.items(): + require(absolute_path(str(path), kind='file') == path and valid_sha(expected) and digest(path) == expected, + 'Pinned file changed: ' + str(path)) + + +def _put(pins, path, sha): + require(valid_sha(sha) and (path not in pins or pins[path] == sha), 'Invalid/conflicting file SHA: ' + str(path)) + pins[path] = sha + + +def _required(pins, path, expected=None): + path = absolute_path(str(path), kind='file') + require(path in pins and (expected is None or pins[path] == expected), 'Required external input pin missing: ' + str(path)) + return path + + +def validate_job(job_file, expected_job_sha256, mode): + """Read/verify all bytes and paths before importing any scientific module.""" + require(mode in ('brain', 'update'), 'Unknown worker mode') + job_file = absolute_path(str(job_file), kind='file') + require(valid_sha(expected_job_sha256) and digest(job_file) == expected_job_sha256, 'External job SHA mismatch') + job = read_json(job_file) + require(type(job) is dict and set(job) == COMMON | (BRAIN_FIELDS if mode == 'brain' else UPDATE_FIELDS), + 'Unknown or missing job fields') + require(job['schema'] == SCHEMA and job['kind'] == mode, 'Job mode/schema mismatch') + require(type(job['job_id']) is str and re.fullmatch('[A-Za-z0-9][A-Za-z0-9_.-]*', job['job_id']) is not None and + '..' not in job['job_id'], 'Unsafe job identity') + require(valid_sha(job['registration_sha256']) and valid_sha(job['worker_sha256']) and + digest(__file__) == job['worker_sha256'], 'Worker/registration pin invalid') + require(type(job['input_sha256']) is dict and bool(job['input_sha256']), 'Nonempty external input pins required') + pins = {job_file: expected_job_sha256, Path(__file__).resolve(): job['worker_sha256']} + for name, sha in job['input_sha256'].items(): + _put(pins, absolute_path(name, kind='file'), sha) + roles = {'runtime_light', 'runtime_continuation'} if mode == 'brain' else {'conditioning_epoch'} + require(type(job['sources']) is dict and set(job['sources']) == roles, 'Wrong worker source roles') + sources = {} + for role, entry in job['sources'].items(): + require(type(entry) is dict and set(entry) == {'path', 'sha256'} and entry['sha256'] == SOURCE_PINS[role], + 'Frozen worker dependency pin differs: ' + role) + sources[role] = absolute_path(entry['path'], kind='file') + _put(pins, sources[role], entry['sha256']) + output = absolute_path(job['output_dir']) + receipt = absolute_path(job['receipt_file']) + attempt = Path(str(receipt) + '.attempt.json') + require(output.parent.is_dir() and receipt.parent.is_dir(), 'Output/receipt parents must be created by coordinator') + require(not output.exists() and not receipt.exists() and not attempt.exists(), 'Existing output/receipt/attempt; retry forbidden') + paths = dict(job=job_file, output=output, receipt=receipt, attempt=attempt) + if mode == 'brain': + require(job['diagnostic_mode'] in ('light', 'reference'), 'Invalid recording mode') + require(job['admission'] == {'gpu_uuid': GPU, 'comfy_queue_urls': list(COMFY_URLS)}, 'GPU/Comfy admission differs') + case = absolute_path(job['case_dir'], kind='dir') + spec_path = _required(pins, case / 'spec.json') + _required(pins, case / 'requests.npz') + spec = read_json(spec_path) + overlay = _required(pins, job['overlay_file']) + manifest = _required(pins, job['overlay_manifest']) + meta = read_json(manifest) + for role in ('mask', 'base_candidate', 'am_overlay'): + _required(pins, meta[role + '_path'], meta[role + '_sha256']) + graph, apl = (absolute_path(spec[key], kind='dir') for key in ('graph_dir', 'apl_bundle_dir')) + for name in GRAPH_FILES: + _required(pins, graph / name) + for name in APL_FILES: + _required(pins, apl / name) + _required(pins, spec['coefficients_path']) + require(sources['runtime_continuation'].parent / 'runtime_light.py' == sources['runtime_light'], + 'Checkpoint reader must use declared adjacent runtime') + build = absolute_path(job['build_root']) + require(not build.exists() or build.is_dir(), 'Build cache is not a directory') + require(build.parent.is_dir(), 'Build parent must already exist') + paths.update(case=case, overlay=overlay, manifest=manifest, build=build) + else: + authority = _required(pins, job['authority_file'], job['authority_sha256']) + value = read_json(authority) + require(type(value.get('files')) is dict and type(value.get('frozen_inputs')) is dict, 'Authority input inventory missing') + for entry in value['files'].values(): + require(type(entry) is dict and set(entry) == {'path', 'sha256'}, 'Invalid authority input row') + _required(pins, entry['path'], entry['sha256']) + for name, sha in value['frozen_inputs'].items(): + _required(pins, name, sha) + checkpoint = absolute_path(value['checkpoint_dir'], kind='dir') + _required(pins, checkpoint / 'checkpoint-manifest.json', value['checkpoint_manifest_sha256']) + paths.update(authority=authority, checkpoint=checkpoint) + protected = {path.parent for path in pins} + if mode == 'update': + protected.add(paths['checkpoint']) + for target in (output, receipt, attempt, *([paths['build']] if mode == 'brain' else [])): + require(not any(target.is_relative_to(root) for root in protected), 'Output/receipt/cache inside protected input/source directory') + require(not receipt.is_relative_to(output) and not output.is_relative_to(receipt), 'Receipt must remain outside sealed output') + if mode == 'brain': + build = paths['build'] + require(not output.is_relative_to(build) and not build.is_relative_to(output) and not receipt.is_relative_to(build), + 'Build cache/output/receipt scopes overlap') + verify_pins(pins) + return job, paths, pins, sources + + +def _load_source(name, path, expected): + """Execute exactly verified bytes; never reuse/repair an existing binding.""" + raw = path.read_bytes() + require(hashlib.sha256(raw).hexdigest() == expected and name not in sys.modules, 'Unverified source/module binding: ' + name) + spec = importlib.util.spec_from_file_location(name, path) + require(spec is not None and spec.loader is not None, 'No Python source loader') + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module # Normal verified import registration, not cache repair. + exec(compile(raw, str(path), 'exec'), module.__dict__) + return module + + +def _clean_process(mode): + require(sys.dont_write_bytecode, 'Use Python -B') + forbidden = ('torch', 'runtime_light', 'runtime_continuation', 'conditioning_epoch', + 'conditioning_preflight_workflow', 'conditioning_preflight_prepare', 'engineering_workflow', + 'phase4an_continuation_pinned_light', 'phase4an_sealed_am_overlay', + '_conditioning_epoch_continuation', '_phase4ap_epoch', '_phase4ap_checkpoint') + require(not any(name in sys.modules for name in forbidden), 'Worker requires a fresh exec interpreter, not a shared/forked runtime') + os.environ['CUDA_VISIBLE_DEVICES'] = '' if mode == 'update' else GPU + if mode == 'brain': + os.environ['CUDA_HOME'] = '/usr/local/cuda' + os.environ['MAX_JOBS'] = '2' + os.environ['TORCH_CUDA_ARCH_LIST'] = '8.9' + os.environ['PATH'] = SCIENCE_BIN + os.pathsep + '/usr/local/cuda/bin' + os.pathsep + os.environ.get('PATH', '') + + +def _gpu_inventory(): + result = subprocess.run(['nvidia-smi', '-i', GPU, + '--query-gpu=uuid,name,memory.total,driver_version', '--format=csv,noheader'], + check=True, capture_output=True, text=True, timeout=20).stdout.strip() + require(len(result.splitlines()) == 1 and result.startswith(GPU + ',') and '4090' in result, 'Wrong physical GPU inventory') + return result + + +def _admit_brain(): + rows = [] + for url in COMFY_URLS: + with urllib.request.urlopen(url, timeout=5) as response: + require(response.geturl() == url, 'Comfy admission redirect forbidden') + raw = response.read(2 * 1024 * 1024 + 1) + require(len(raw) <= 2 * 1024 * 1024, 'Oversized Comfy admission response') + queue = json.loads(raw) + require(type(queue) is dict and queue.get('queue_running') == [] and queue.get('queue_pending') == [], + 'ComfyUI busy or malformed; never evict a service') + rows.append(dict(url=url, queue_running=0, queue_pending=0, response_sha256=hashlib.sha256(raw).hexdigest())) + return dict(gpu_uuid=GPU, nvidia_smi_inventory=_gpu_inventory(), comfy_queues=rows, + observed_utc=datetime.now(timezone.utc).isoformat(), check_is_not_a_service_lock=True) + + +def _environment(mode): + result = dict(python_executable=sys.executable, python_version=sys.version, + python_no_bytecode=sys.dont_write_bytecode, PATH=os.environ.get('PATH'), + environment_scope='selected execution variables and library/precision metadata; not a dump of credentials', + CUDA_VISIBLE_DEVICES=os.environ.get('CUDA_VISIBLE_DEVICES'), + CUBLAS_WORKSPACE_CONFIG=os.environ.get('CUBLAS_WORKSPACE_CONFIG'), + TORCH_CUDA_ARCH_LIST=os.environ.get('TORCH_CUDA_ARCH_LIST'), + CUDA_HOME=os.environ.get('CUDA_HOME'), MAX_JOBS=os.environ.get('MAX_JOBS'), + settings_changed_by_environment_recorder=False) + torch, numpy = sys.modules.get('torch'), sys.modules.get('numpy') + result.update(numpy_version=None if numpy is None else numpy.__version__, + torch_version=None if torch is None else str(torch.__version__)) + if mode == 'update': + require(os.environ.get('CUDA_VISIBLE_DEVICES') == '' and (torch is None or not torch.cuda.is_initialized()), + 'Update worker initialized or exposed CUDA') + result.update(gpu_uuid=None, CUDA_hidden=True, CUDA_initialized=False, + device_scope='CPU eligibility only; no brain execution') + return result + require(torch is not None and os.environ.get('CUDA_VISIBLE_DEVICES') == GPU and torch.cuda.device_count() == 1, + 'Brain runtime is not using the single admitted GPU') + properties = torch.cuda.get_device_properties(0) + require('4090' in properties.name, 'Unexpected GPU device class') + result.update(torch_cuda_version=torch.version.cuda, device_name=properties.name, + total_memory_bytes=int(properties.total_memory), gpu_uuid=GPU, nvidia_smi_inventory=_gpu_inventory(), + matmul_allow_tf32=bool(torch.backends.cuda.matmul.allow_tf32), cudnn_allow_tf32=bool(torch.backends.cudnn.allow_tf32), + deterministic_algorithms=bool(torch.are_deterministic_algorithms_enabled()), + cudnn_deterministic=bool(torch.backends.cudnn.deterministic), cudnn_benchmark=bool(torch.backends.cudnn.benchmark), + float32_matmul_precision=torch.get_float32_matmul_precision(), environment_recorder_adds_no_timing=True) + return result + + +def _verify_precision(environment, component): + require(component.get('cuda_matmul_allow_tf32') is False and component.get('float32_matmul_precision') == 'highest' and + component.get('precision_configuration_changed_by_component') is False, 'Frozen precision configuration differs') + pairs = {'matmul_allow_tf32': 'cuda_matmul_allow_tf32', 'cudnn_allow_tf32': 'cudnn_allow_tf32', + 'deterministic_algorithms': 'deterministic_algorithms', 'cudnn_deterministic': 'cudnn_deterministic', + 'cudnn_benchmark': 'cudnn_benchmark', 'float32_matmul_precision': 'float32_matmul_precision', + 'CUBLAS_WORKSPACE_CONFIG': 'CUBLAS_WORKSPACE_CONFIG', 'device_name': 'device_name', + 'torch_version': 'torch', 'torch_cuda_version': 'cuda', 'numpy_version': 'numpy'} + require(all(left in environment and right in component and + type(environment[left]) is type(component[right]) and environment[left] == component[right] + for left, right in pairs.items()), 'Actual environment and APL component disagree') + + +def _run_brain(job, paths, pins, sources): + r = _load_source('runtime_light', sources['runtime_light'], SOURCE_PINS['runtime_light']) + meta = read_json(paths['manifest']) + for relative, sha in r.PINS.items(): + _required(pins, Path(meta['aj_model_dir']).parent / relative, sha) + r.execution_paths(paths['case'], paths['output'], paths['build'], protected={p.parent for p in pins}) + admission = _admit_brain() + execute = r.run_reference if job['diagnostic_mode'] == 'reference' else r.run_light + summary = execute(paths['case'], paths['output'], paths['build'], paths['overlay'], paths['manifest']) + # This reader uses a separate pinned R but does NOT load the shared helper. + # E must never be loaded in this process; no caches are patched or cleared. + c = _load_source('_phase4ap_checkpoint', sources['runtime_continuation'], SOURCE_PINS['runtime_continuation']) + seal = c.seal_checkpoint(paths['output']) + record = c.load_checkpoint(paths['output'], seal['sha256']) + require(record['fresh'] is True and record['global_tick_offset'] == 0 and + record['summary'] == summary and summary['diagnostic_mode'] == job['diagnostic_mode'], 'Sealed runtime scope differs') + environment = _environment('brain') + _verify_precision(environment, summary['apl_component']['environment']) + return dict(checkpoint_manifest_sha256=seal['sha256'], checkpoint_manifest_file=str(paths['output'] / 'checkpoint-manifest.json'), + complete_fresh_trial=True, steps=record['steps'], admission=admission, environment=environment, + within_trial_weight_updates=False, epoch_update_performed=False) + + +def _run_update(job, paths, pins, sources): + e = _load_source('_phase4ap_epoch', sources['conditioning_epoch'], SOURCE_PINS['conditioning_epoch']) + chain = e.postprocess_epoch(paths['authority'], job['authority_sha256'], paths['output']) + require(chain.get('complete') is True and chain.get('authority_sha256') == job['authority_sha256'] and + chain.get('weights_installed') is False and chain.get('brain_executed') is False and + chain.get('learning_pass_claimed') is False, 'Incomplete/incorrect epoch postprocessing scope') + require(chain == read_json(paths['output'] / 'chain-receipt.json'), 'Stored chain differs from actual result') + require('runtime_light' not in sys.modules and 'runtime_continuation' not in sys.modules and + 'conditioning_preflight_workflow' not in sys.modules, 'Update imported the mixed-runtime coordinator') + return dict(chain_receipt_sha256=digest(paths['output'] / 'chain-receipt.json'), + source_neural_checkpoint_manifest_sha256=chain['source_neural_checkpoint_manifest_sha256'], + before_overlay_file_sha256=chain['before_overlay_file_sha256'], after_overlay_file_sha256=chain['after_overlay_file_sha256'], + before_overlay_manifest_sha256=chain['before_overlay_manifest_sha256'], after_overlay_manifest_sha256=chain['after_overlay_manifest_sha256'], + environment=_environment('update'), brain_executed=False, weights_installed=False, + registration_authority_independently_proven=False) + + +def tree_hashes(directory): + require(directory.is_dir(), 'Worker produced no output directory') + result = {} + for path in sorted(directory.rglob('*')): + require(not path.is_symlink() and path.resolve().is_relative_to(directory), 'Unsafe worker output tree') + if path.is_dir(): + continue + require(path.is_file(), 'Nonregular worker output') + result[path.relative_to(directory).as_posix()] = digest(path) + require(bool(result), 'Empty worker output') + return result + + +def execute_job(mode, job_file, expected_job_sha256): + """CLI implementation. Production never substitutes dependencies or arrays.""" + _clean_process(mode) + job, paths, pins, sources = validate_job(job_file, expected_job_sha256, mode) + started = time.perf_counter() + save_json(paths['attempt'], dict(schema='phase4ap-worker-attempt-v1', complete=False, + job_sha256=expected_job_sha256, job_id=job['job_id'], kind=mode, pid=os.getpid(), + started_utc=datetime.now(timezone.utc).isoformat(), no_automatic_retry=True)) + result = (_run_brain if mode == 'brain' else _run_update)(job, paths, pins, sources) + outputs = tree_hashes(paths['output']) + verify_pins(pins) + require(outputs == tree_hashes(paths['output']), 'Worker output changed during receipt creation') + receipt = dict(schema='phase4ap-isolated-worker-receipt-v1', complete=True, + kind=mode, job_id=job['job_id'], job_file=str(paths['job']), job_sha256=expected_job_sha256, + registration_sha256=job['registration_sha256'], worker_sha256=job['worker_sha256'], + pid=os.getpid(), argv=[sys.executable, *sys.argv], wall_seconds=time.perf_counter() - started, + attempt_sha256=digest(paths['attempt']), source_files=job['sources'], + verified_input_sha256={str(path): sha for path, sha in sorted(pins.items(), key=lambda item: str(item[0]))}, + output_dir=str(paths['output']), receipt_file=str(paths['receipt']), output_files_sha256=outputs, result=result, + environment=result['environment'], completion_semantics='worker returned successfully before writing this receipt; parent must independently require actual process exit 0', + process_exit_independently_observed=False, registration_authority_independently_proven=False, + learning_success_claimed=False, scientific_gate_pass_claimed=False, + old_sources_modified=False, loader_caches_patched=False) + save_json(paths['receipt'], receipt) + return receipt + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('mode', choices=('brain', 'update')) + parser.add_argument('--job', required=True) + parser.add_argument('--job-sha256', required=True) + args = parser.parse_args(argv) + receipt = execute_job(args.mode, args.job, args.job_sha256) + print(json.dumps(dict(complete=True, kind=args.mode, job_id=receipt['job_id'], + receipt_file=receipt['receipt_file'], + receipt_sha256=digest(receipt['receipt_file'])), sort_keys=True), flush=True) + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/models/male-cns-lif/conditioning-interface-v2/test_coordinator.py b/models/male-cns-lif/conditioning-interface-v2/test_coordinator.py new file mode 100644 index 0000000..1537d85 --- /dev/null +++ b/models/male-cns-lif/conditioning-interface-v2/test_coordinator.py @@ -0,0 +1,202 @@ +"""Synthetic CPU contract tests, not a neural/scientific PASS claim.""" +import ast +import copy +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +import coordinator as M + + +class CoordinatorTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.exp = Path(self.tmp.name) + self.patch = patch.object(M, 'EXP', self.exp) + self.patch.start(); self.addCleanup(self.patch.stop) + for name in ('jobs','receipts','logs','audit','updates','runs','cases'): + (self.exp/name).mkdir() + self.row = dict(row_id='D1--reference--epoch-001', arm='D1', diagnostic_mode='reference', + input_epoch_index=1, previous_delta_row_id='D1--reference--epoch-000', + origin='new_AP_neural_execution', case_dir=str(self.exp/'cases/one'), + neural_dir=str(self.exp/'runs/one'), steps=1200) + + def write(self, path, value): + path.write_text(__import__('json').dumps(value), encoding='utf8') + + def fixture(self): + """Explicit synthetic files only; no old paths or runtime payload.""" + regsha='1'*64; loc=M.process_locations(self.row,'update') + output=M.paths(self.row)['update'];output.mkdir() + worker_source=M.MODEL/'isolated_workers.py' + authority=loc['job'].parent/'one.authority.json';M.save(authority,{'fixture':True}) + job=dict(job_id=self.row['row_id']+'.update', kind='update',registration_sha256=regsha, + worker_sha256=M.digest(worker_source),output_dir=str(output),receipt_file=str(loc['worker']), + sources={},input_sha256={str(authority):M.digest(authority)},authority_sha256=M.digest(authority)) + M.save(loc['job'],job) + (output/'delta-after.npz').write_bytes(b'SYNTHETIC-NOT-NPZ') + M.save(output/'overlay-manifest.json',{'fixture':True}) + chain=dict(authority_sha256=M.digest(authority),source_neural_checkpoint_manifest_sha256='2'*64, + before_overlay_file_sha256='3'*64,before_overlay_manifest_sha256='4'*64, + after_overlay_file_sha256=M.digest(output/'delta-after.npz'), + after_overlay_manifest_sha256=M.digest(output/'overlay-manifest.json')) + M.save(output/'chain-receipt.json',chain) + attempt=Path(str(loc['worker'])+'.attempt.json');M.save(attempt,{'fixture':True}) + result={**chain,'chain_receipt_sha256':M.digest(output/'chain-receipt.json'), + 'brain_executed':False,'weights_installed':False} + result.pop('authority_sha256') + worker=dict(schema='phase4ap-isolated-worker-receipt-v1',complete=True, + job_sha256=M.digest(loc['job']),registration_sha256=regsha,job_id=job['job_id'],kind='update', + worker_sha256=job['worker_sha256'],output_dir=str(output),receipt_file=str(loc['worker']), + source_files={},attempt_sha256=M.digest(attempt), + verified_input_sha256={**job['input_sha256'],str(loc['job']):M.digest(loc['job']),str(worker_source):M.digest(worker_source)}, + output_files_sha256={p.name:M.digest(p) for p in output.iterdir()},result=result, + environment={'CUDA_hidden':True,'CUDA_initialized':False}) + M.save(loc['worker'],worker) + loc['stdout'].write_bytes(b'');loc['stderr'].write_bytes(b'') + process=dict(complete=True,exit_code=0,registration_sha256=regsha,row_id=self.row['row_id'],kind='update', + job_sha256=M.digest(loc['job']),worker_receipt_sha256=M.digest(loc['worker']), + stdout_sha256=M.digest(loc['stdout']),stderr_sha256=M.digest(loc['stderr'])) + M.save(loc['process'],process) + return regsha,loc,job,worker,process + + def test_exact_process_fixture(self): + sha,loc,job,worker,process=self.fixture() + self.assertEqual(M.verify_process({},sha,self.row,'update',job),worker) + + def test_mutated_receipt_fields_rejected_even_if_process_rehashed(self): + sha,loc,job,worker,process=self.fixture() + mutations=[('output_files_sha256',{}),('verified_input_sha256',{}),('kind','brain'), + ('output_dir',str(self.exp/'wrong')),('source_files',{'wrong':1}),('attempt_sha256','0'*64), + ('worker_sha256','0'*64),('receipt_file',str(self.exp/'fake'))] + for field,value in mutations: + with self.subTest(field=field): + changed={**worker,field:value};self.write(loc['worker'],changed) + self.write(loc['process'],{**process,'worker_receipt_sha256':M.digest(loc['worker'])}) + with self.assertRaises(ValueError):M.verify_process({},sha,self.row,'update',job) + + def test_result_chain_mutation_rejected(self): + sha,loc,job,worker,process=self.fixture() + for field in ('chain_receipt_sha256','before_overlay_file_sha256','after_overlay_file_sha256', + 'source_neural_checkpoint_manifest_sha256'): + changed=copy.deepcopy(worker);changed['result'][field]='0'*64 + self.write(loc['worker'],changed) + self.write(loc['process'],{**process,'worker_receipt_sha256':M.digest(loc['worker'])}) + with self.assertRaises(ValueError):M.verify_process({},sha,self.row,'update',job) + + def test_output_bytes_and_extra_files_rejected(self): + sha,loc,job,worker,process=self.fixture() + (Path(job['output_dir'])/'unregistered').write_bytes(b'new') + with self.assertRaisesRegex(ValueError,'inventory'):M.verify_process({},sha,self.row,'update',job) + + def test_job_changes_rejected(self): + sha,loc,job,worker,process=self.fixture() + with self.assertRaisesRegex(ValueError,'Job differs'): + M.verify_process({},sha,self.row,'update',{**job,'job_id':'another'}) + + def test_spec_only_budget_changes(self): + original={'budget_root':str(self.exp/'old'),'output_budget_bytes':50,'steps':1200,'seed':5717173809506784725, + 'lif':{'dt':.1,'delay':18},'other':[1,2]} + result,report=M.derive_spec(original,str(self.exp/'new'),100) + self.assertEqual(result['seed'],5717173809506784725) + self.assertEqual(set(report['changes']),{'budget_root','output_budget_bytes'}) + result['lif']['dt']=10 + self.assertEqual(original['lif']['dt'],.1) + with self.assertRaises(ValueError):M.derive_spec(original,original['budget_root'],100) + + def test_rows_exact_unique_and_safe_paths(self): + self.assertEqual(M.slot({'slots':[self.row]},self.row['row_id']),self.row) + for rows in ([],[self.row,self.row]): + with self.assertRaises(ValueError):M.slot({'slots':rows},self.row['row_id']) + for value in ('../x','a/b','a\\b','',None): + with self.assertRaises(ValueError):M.paths({**self.row,'row_id':value}) + + def test_parent_arm_mode_epoch_cannot_cross(self): + parent={**self.row,'row_id':self.row['previous_delta_row_id'],'input_epoch_index':0} + for change in ({'arm':'D0'},{'diagnostic_mode':'light'},{'input_epoch_index':2}): + with self.assertRaisesRegex(ValueError,'ancestry'): + M.before_overlay({'slots':[{**parent,**change}]},'unused',self.row) + + def test_parent_verified_before_delta_is_selected(self): + parent={**self.row,'row_id':self.row['previous_delta_row_id'],'input_epoch_index':0} + with patch.object(M,'verify_slot',side_effect=ValueError('parent invalid')) as verify: + with self.assertRaisesRegex(ValueError,'parent invalid'): + M.before_overlay({'slots':[parent]},'unused',self.row) + verify.assert_called_once() + + def test_initial_overlay_has_no_previous_read(self): + initial={'overlay_file':'immutable'} + with patch.object(M,'verify_slot') as verify: + self.assertIs(M.before_overlay({'initial_overlay':initial},'unused',{'previous_delta_row_id':None}),initial) + verify.assert_not_called() + + def test_imported_neural_cannot_execute(self): + worker=str(M.MODEL/'isolated_workers.py') + reg={'input_sha256':{},'source_sha256':{worker:'1'*64},'prepared_file_sha256':{}} + with self.assertRaisesRegex(ValueError,'Imported neural'): + M.job_for(reg,'sha',{**self.row,'origin':'imported_sealed_AO_neural_trajectory'},'brain',{}) + with self.assertRaisesRegex(ValueError,'Unknown child'): + M.job_for({},'sha',self.row,'other') + + def test_stopped_run_and_budget_refuse_without_delete(self): + reg={'budget':{'output_budget_bytes':10,'minimum_free_reserve_bytes':1}} + marker=self.exp/'audit/stopped.json';marker.write_bytes(b'{}') + with self.assertRaisesRegex(ValueError,'Stopped/completed'):M.admission(reg) + self.assertTrue(marker.exists()) + + def test_partial_child_not_retried(self): + loc=M.process_locations(self.row,'update');loc['stdout'].write_bytes(b'partial') + with patch.object(M.subprocess,'run') as run: + with self.assertRaisesRegex(ValueError,'cannot be retried'):M.launch({},'sha',self.row,'update',{}) + run.assert_not_called() + + def test_failed_child_records_actual_failure(self): + with patch.object(M,'admission'),patch.object(M.subprocess,'run',return_value=subprocess.CompletedProcess([],7)): + with self.assertRaisesRegex(ValueError,'failed or timed out'): + M.launch({'budget':{'child_timeout_seconds':1}},'sha',self.row,'update',{'fixture':True}) + record=M.read(M.process_locations(self.row,'update')['process']) + self.assertFalse(record['complete']);self.assertEqual(record['exit_code'],7) + + def test_timedout_child_never_claims_exit_zero(self): + with patch.object(M,'admission'),patch.object(M.subprocess,'run',side_effect=subprocess.TimeoutExpired([],1)): + with self.assertRaises(ValueError): + M.launch({'budget':{'child_timeout_seconds':1}},'sha',self.row,'update',{'fixture':True}) + record=M.read(M.process_locations(self.row,'update')['process']) + self.assertTrue(record['timed_out']);self.assertIsNone(record['exit_code']) + + def test_wrong_registration_sha_stops_before_parse(self): + (self.exp/'protocol').mkdir();M.save(self.exp/'protocol/preregistered.json',{'fake':True}) + with self.assertRaisesRegex(ValueError,'external SHA'):M.registration('0'*64) + + def test_no_postprocess_or_sampler_call_in_coordinator(self): + tree=ast.parse(Path(M.__file__).read_text(encoding='utf8')) + calls=[ast.unparse(n.func) for n in ast.walk(tree) if isinstance(n,ast.Call)] + self.assertFalse(any('postprocess_epoch' in n or 'generate_epoch_inputs' in n for n in calls)) + self.assertNotIn('os.fork',calls) + + def test_actual_delta_must_match_correct_receipt(self): + from test_conditioning_preflight_compare import delta_fixture + before,after,installed,receipt,manifest=delta_fixture() + overlay={'overlay_file':'synthetic'} + with patch.object(M.R,'load_overlay',return_value=(None,{'plastic_delta_mv':after},None,None)), \ + patch.object(M,'npz',return_value={'plastic_delta_mv':before}), \ + patch.object(M,'digest',return_value='3'*64),patch.object(M,'read',return_value=manifest): + self.assertTrue(M.verify_delta_payload(overlay,self.exp,receipt)[0]['pass_all']) + after[1]=0 + with self.assertRaisesRegex(ValueError,'Actual delta binding'): + M.verify_delta_payload(overlay,self.exp,receipt) + + def test_imported_comparison_uses_bits_and_original_receipt(self): + import numpy as np + value=np.zeros(3,np.float32);receipt={'real':'fixture'} + with patch.object(M,'npz',return_value={'plastic_delta_mv':value}),patch.object(M,'read',return_value=receipt): + M.verify_imported_delta(value.copy(),copy.deepcopy(receipt)) + changed=value.copy();changed[0]=np.float32(-0.) + with self.assertRaisesRegex(ValueError,'Imported trajectory'):M.verify_imported_delta(changed,receipt) + with self.assertRaises(ValueError):M.verify_imported_delta(value,{'different':True}) + + +if __name__=='__main__':unittest.main() diff --git a/models/male-cns-lif/conditioning-interface-v2/test_isolated_workers.py b/models/male-cns-lif/conditioning-interface-v2/test_isolated_workers.py new file mode 100644 index 0000000..362d469 --- /dev/null +++ b/models/male-cns-lif/conditioning-interface-v2/test_isolated_workers.py @@ -0,0 +1,346 @@ +"""Synthetic/read-only-loader tests only: never invoke GPU or real events.""" +import ast +import copy +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +from types import ModuleType, SimpleNamespace +import unittest +from unittest.mock import patch + +import isolated_workers as w + + +LAB = Path(__file__).resolve().parent.parent / 'olfactory-conditioning-lab-v1' +AM = LAB.parent / 'wholebrain-plastic-interface-v1/overlay.py' + + +def write_json(path, value): + path.write_text(json.dumps(value, sort_keys=True, allow_nan=False), encoding='utf-8') + + +class Fixture: + def __init__(self, root, mode): + self.root = root + for name in ('data', 'jobs', 'outputs', 'receipts', 'cache'): + (root / name).mkdir() + self.data, self.job_file = root / 'data', root / 'jobs/job.json' + for name in ('spec.json', 'requests.npz', 'before.npz', 'before.json', 'mask.npy', 'base.npy', 'coeff.json'): + (self.data / name).write_bytes(b'fixture bytes, not a scientific array\n') + self.output, self.receipt = root / 'outputs/new', root / 'receipts/new.json' + source_roles = ('runtime_light', 'runtime_continuation') if mode == 'brain' else ('conditioning_epoch',) + self.job = dict(schema=w.SCHEMA, job_id='synthetic-row.' + mode, kind=mode, registration_sha256='a' * 64, + worker_sha256=w.digest(w.__file__), output_dir=str(self.output), receipt_file=str(self.receipt), + sources={role: dict(path=str(LAB / (role + '.py')), sha256=w.SOURCE_PINS[role]) for role in source_roles}, + input_sha256={}) + if mode == 'brain': + graph, apl = self.data / 'graph', self.data / 'apl' + graph.mkdir(); apl.mkdir() + for name in w.GRAPH_FILES: + (graph / name).write_bytes(b'not a real graph\n') + for name in w.APL_FILES: + (apl / name).write_bytes(b'not a real APL array\n') + write_json(self.data / 'spec.json', dict(graph_dir=str(graph), apl_bundle_dir=str(apl), + coefficients_path=str(self.data / 'coeff.json'))) + write_json(self.data / 'before.json', dict(mask_path=str(self.data / 'mask.npy'), + mask_sha256=w.digest(self.data / 'mask.npy'), base_candidate_path=str(self.data / 'base.npy'), + base_candidate_sha256=w.digest(self.data / 'base.npy'), am_overlay_path=str(AM), + am_overlay_sha256=w.digest(AM))) + self.job.update(case_dir=str(self.data), build_root=str(root / 'cache/build'), + overlay_file=str(self.data / 'before.npz'), overlay_manifest=str(self.data / 'before.json'), + diagnostic_mode='reference', admission=dict(gpu_uuid=w.GPU, comfy_queue_urls=list(w.COMFY_URLS))) + self.job['input_sha256'][str(AM)] = w.digest(AM) + else: + (self.data / 'checkpoint').mkdir() + (self.data / 'checkpoint/checkpoint-manifest.json').write_text('{}', encoding='utf-8') + write_json(self.data / 'authority.json', dict(files={'spec': dict(path=str(self.data / 'spec.json'), + sha256=w.digest(self.data / 'spec.json'))}, frozen_inputs={str(self.data / 'before.npz'): w.digest(self.data / 'before.npz')}, + checkpoint_dir=str(self.data / 'checkpoint'), + checkpoint_manifest_sha256=w.digest(self.data / 'checkpoint/checkpoint-manifest.json'))) + self.job.update(authority_file=str(self.data / 'authority.json'), authority_sha256=w.digest(self.data / 'authority.json')) + self.job['input_sha256'].update({str(p): w.digest(p) for p in self.data.rglob('*') if p.is_file()}) + self.save() + + def save(self): + write_json(self.job_file, self.job) + return w.digest(self.job_file) + + def validate(self): + return w.validate_job(self.job_file, w.digest(self.job_file), self.job['kind']) + + +class JobTests(unittest.TestCase): + def fixture(self, mode='update'): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + return Fixture(Path(tmp.name), mode) + + def test_top_level_stdlib_only_and_hard_pins_actual(self): + tree = ast.parse(Path(w.__file__).read_text(encoding='utf-8')) + for node in tree.body: + if isinstance(node, ast.Import): + names = [item.name.split('.')[0] for item in node.names] + elif isinstance(node, ast.ImportFrom): + names = [node.module.split('.')[0]] + else: + continue + self.assertTrue(set(names) <= sys.stdlib_module_names | {'__future__'}, names) + for name, expected in w.SOURCE_PINS.items(): + self.assertEqual(w.digest(LAB / (name + '.py')), expected) + + def test_valid_job_schema_for_both_modes_no_execution(self): + for mode in ('brain', 'update'): + with self.subTest(mode=mode): + f = self.fixture(mode) + job, paths, pins, sources = f.validate() + self.assertEqual(job['kind'], mode) + self.assertEqual(paths['output'], f.output) + self.assertIn(f.job_file, pins) + self.assertEqual(set(sources), {'runtime_light', 'runtime_continuation'} if mode == 'brain' else {'conditioning_epoch'}) + self.assertFalse(f.output.exists()) + self.assertFalse(f.receipt.exists()) + + def test_external_job_hash_and_file_mutation(self): + f = self.fixture() + original = w.digest(f.job_file) + f.job['job_id'] = 'changed'; f.save() + with self.assertRaisesRegex(ValueError, 'job SHA'): + w.validate_job(f.job_file, original, 'update') + (f.data / 'spec.json').write_bytes(b'mutation') + with self.assertRaisesRegex(ValueError, 'Pinned file changed'): + f.validate() + self.assertFalse(f.output.exists()) + + def test_unknown_mode_bool_unknown_fields_and_bad_source(self): + for mutation in ( + lambda j: j.update(kind=True), lambda j: j.update(extra=True), + lambda j: j.update(job_id='../escape'), lambda j: j.update(job_id='a..b'), + lambda j: j.update(authority_sha256=True), + lambda j: j['sources']['conditioning_epoch'].update(sha256='f' * 64), + lambda j: j['sources'].update(runtime_light=dict(path=str(LAB / 'runtime_light.py'), sha256=w.SOURCE_PINS['runtime_light'])), + ): + f = self.fixture(); mutation(f.job); f.save() + with self.assertRaises((ValueError, KeyError)): + w.validate_job(f.job_file, w.digest(f.job_file), 'update') + + def test_same_sha_does_not_authorize_relocated_input(self): + f = self.fixture() + twin = f.root / 'jobs/twin.json' + twin.write_bytes((f.data / 'authority.json').read_bytes()) + f.job['authority_file'] = str(twin); f.save() + with self.assertRaisesRegex(ValueError, 'Required external input pin missing'): + f.validate() + + def test_required_graph_and_authority_pins_cannot_be_omitted(self): + f = self.fixture('brain') + del f.job['input_sha256'][str(f.data / 'graph/ptr.npy')]; f.save() + with self.assertRaisesRegex(ValueError, 'Required external input pin missing'): + f.validate() + f = self.fixture() + del f.job['input_sha256'][str(f.data / 'spec.json')]; f.save() + with self.assertRaisesRegex(ValueError, 'Required external input pin missing'): + f.validate() + + def test_receipt_output_cache_paths_are_exclusive_and_protected(self): + cases = [('output_dir', 'data/new'), ('receipt_file', 'data/new.json'), + ('receipt_file', 'outputs/new/receipt.json'), ('build_root', 'data/cache'), + ('build_root', 'outputs/new/cache')] + for key, suffix in cases: + with self.subTest(key=key, suffix=suffix): + f = self.fixture('brain') + f.job[key] = str(f.root / suffix); f.save() + with self.assertRaises(ValueError): + f.validate() + for what in ('output', 'receipt', 'attempt'): + f = self.fixture() + target = {'output': f.output, 'receipt': f.receipt, 'attempt': Path(str(f.receipt) + '.attempt.json')}[what] + if what == 'output': target.mkdir() + else: target.write_text('existing') + with self.assertRaisesRegex(ValueError, 'retry forbidden'): + f.validate() + + def test_path_link_escape_rejected(self): + f = self.fixture() + link = f.root / 'linked' + try: + link.symlink_to(f.data, target_is_directory=True) + except OSError as error: + self.skipTest('Local symlink privilege unavailable: ' + str(error)) + f.job['output_dir'] = str(link / 'new'); f.save() + with self.assertRaisesRegex(ValueError, 'Linked/junction'): + f.validate() + + def test_wrong_gpu_or_missing_comfy_admission_rejected(self): + for mutation in (lambda j: j['admission'].update(gpu_uuid='GPU-other'), + lambda j: j['admission'].update(comfy_queue_urls=[])): + f = self.fixture('brain'); mutation(f.job); f.save() + with self.assertRaisesRegex(ValueError, 'admission differs'): + f.validate() + + def test_update_hides_cuda_before_scientific_import(self): + # The unittest host may legitimately have other suites' runtimes + # imported. The production contract is a fresh exec child, not that + # host; exercise the real boundary instead of clearing host modules. + code = "import os,sys; import isolated_workers as w; os.environ['CUDA_VISIBLE_DEVICES']=w.GPU; w._clean_process('update'); assert os.environ['CUDA_VISIBLE_DEVICES']==''; e=w._environment('update'); assert e['CUDA_hidden'] and not e['CUDA_initialized']; assert 'torch' not in sys.modules; print('hidden-before-import')" + result = subprocess.run([sys.executable, '-B', '-c', code], cwd=Path(w.__file__).parent, + capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('hidden-before-import', result.stdout) + + def test_fake_preimport_is_rejected_not_removed(self): + bad = ModuleType('phase4an_sealed_am_overlay') + with patch.dict(sys.modules, {'phase4an_sealed_am_overlay': bad}): + with self.assertRaisesRegex(ValueError, 'fresh exec'): + w._clean_process('update') + self.assertIs(sys.modules['phase4an_sealed_am_overlay'], bad) + + def test_brain_configures_only_admitted_cuda_and_original_venv_path(self): + code = "import os,sys; import isolated_workers as w; exe=sys.executable; w._clean_process('brain'); assert sys.executable==exe; assert os.environ['CUDA_VISIBLE_DEVICES']==w.GPU; assert os.environ['PATH'].split(os.pathsep)[:2]==[w.SCIENCE_BIN,'/usr/local/cuda/bin']; assert os.environ['TORCH_CUDA_ARCH_LIST']=='8.9'; assert 'torch' not in sys.modules; print('brain-env-no-GPU-call')" + result = subprocess.run([sys.executable, '-B', '-c', code], cwd=Path(w.__file__).parent, + capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('brain-env-no-GPU-call', result.stdout) + + def test_failed_or_mutated_worker_never_receipts_and_attempt_blocks_retry(self): + for failure in ('raise', 'mutate'): + f = self.fixture() + def fake(job, paths, pins, sources): + paths['output'].mkdir() + (paths['output'] / 'partial.txt').write_text('preserved') + if failure == 'raise': + raise RuntimeError('synthetic child failure') + (f.data / 'before.npz').write_bytes(b'changed after validation') + return {'environment': {'synthetic': True}} + with patch.object(w, '_clean_process'), patch.object(w, '_run_update', side_effect=fake): + with self.assertRaises((RuntimeError, ValueError)): + w.execute_job('update', f.job_file, w.digest(f.job_file)) + self.assertFalse(f.receipt.exists()) + self.assertTrue((f.output / 'partial.txt').is_file()) + self.assertTrue(Path(str(f.receipt) + '.attempt.json').is_file()) + with self.assertRaisesRegex(ValueError, 'retry forbidden'): + f.validate() + + def test_success_receipt_binds_inputs_outputs_but_not_process_exit(self): + f = self.fixture() + def fake(job, paths, pins, sources): + paths['output'].mkdir() + write_json(paths['output'] / 'synthetic-only.json', {'not_scientific': True}) + return {'environment': {'synthetic': True}} + with patch.object(w, '_clean_process'), patch.object(w, '_run_update', side_effect=fake): + result = w.execute_job('update', f.job_file, w.digest(f.job_file)) + self.assertTrue(result['complete']) + self.assertFalse(result['process_exit_independently_observed']) + self.assertFalse(result['learning_success_claimed']) + self.assertEqual(result, w.read_json(f.receipt)) + self.assertEqual(result['output_files_sha256'], {'synthetic-only.json': w.digest(f.output / 'synthetic-only.json')}) + self.assertEqual(result['verified_input_sha256'][str(f.job_file)], w.digest(f.job_file)) + self.assertNotIn(str(f.receipt), result['output_files_sha256']) + + def test_comfy_busy_and_malformed_fail_without_gpu_query(self): + for data in ({'queue_running': [1], 'queue_pending': []}, {'queue_running': [], 'queue_pending': [1]}, {}): + response = SimpleNamespace(geturl=lambda: w.COMFY_URLS[0], read=lambda n: json.dumps(data).encode()) + class Open: + def __enter__(self): return response + def __exit__(self, *args): return False + with patch.object(w.urllib.request, 'urlopen', return_value=Open()), patch.object(w, '_gpu_inventory') as gpu: + with self.assertRaisesRegex(ValueError, 'ComfyUI busy'): + w._admit_brain() + gpu.assert_not_called() + + def test_actual_source_byte_mutation_rejected_before_execution(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / 'changed.py' + path.write_bytes((LAB / 'conditioning_epoch.py').read_bytes() + b'\n# mutation\n') + with self.assertRaisesRegex(ValueError, 'Unverified source'): + w._load_source('_test_never_exec', path, w.SOURCE_PINS['conditioning_epoch']) + self.assertNotIn('_test_never_exec', sys.modules) + + +class ActualLoaderSubprocessTests(unittest.TestCase): + def run_python(self, code): + result = subprocess.run([sys.executable, '-B', '-c', code], cwd=LAB, + text=True, capture_output=True, timeout=30) + self.assertEqual(result.returncode, 0, result.stdout + '\n' + result.stderr) + return result.stdout + + @staticmethod + def clean_code(): + return """ +import pathlib,sys,json +import conditioning_epoch as E +p=pathlib.Path.cwd() +c=E._load_source(p/'runtime_continuation.py',E.SOURCE_PINS['runtime_continuation_source'],'continuation') +h=c.R._load('phase4an_sealed_am_overlay',p.parent/'wholebrain-plastic-interface-v1/overlay.py',E.SOURCE_PINS['am_overlay_source']) +assert c.R._load('phase4an_sealed_am_overlay',p.parent/'wholebrain-plastic-interface-v1/overlay.py',E.SOURCE_PINS['am_overlay_source']) is h +assert 'runtime_light' not in sys.modules and 'torch' not in sys.modules +print('clean-E-real-helper-PASS') +""" + + def test_real_E_only_loader_reuses_same_helper_without_torch(self): + self.assertIn('clean-E-real-helper-PASS', self.run_python(self.clean_code())) + + def test_old_mixed_runtime_failure_and_true_exec_child_isolation(self): + code = """ +import pathlib,sys,subprocess +import runtime_light as R +import conditioning_epoch as E +p=pathlib.Path.cwd() +h=R._load('phase4an_sealed_am_overlay',p.parent/'wholebrain-plastic-interface-v1/overlay.py',R.AM_OVERLAY_SHA) +c=E._load_source(p/'runtime_continuation.py',E.SOURCE_PINS['runtime_continuation_source'],'continuation') +assert R is not c.R +try: + c.R._load('phase4an_sealed_am_overlay',p.parent/'wholebrain-plastic-interface-v1/overlay.py',R.AM_OVERLAY_SHA) +except ValueError as e: + assert str(e)=='Existing unverified module binding: phase4an_sealed_am_overlay' +else: raise AssertionError('old loader collision not reproduced') +assert sys.modules['phase4an_sealed_am_overlay'] is h +""" + '\nr=subprocess.run([sys.executable,"-B","-c",' + repr(self.clean_code()) + '],capture_output=True,text=True,timeout=20)\nassert r.returncode==0,r.stderr\nprint(r.stdout)' + self.assertIn('clean-E-real-helper-PASS', self.run_python(code)) + + def test_real_same_runtime_loader_does_not_accept_poisoned_helper(self): + code = """ +import pathlib,sys,types +import conditioning_epoch as E +p=pathlib.Path.cwd() +c=E._load_source(p/'runtime_continuation.py',E.SOURCE_PINS['runtime_continuation_source'],'continuation') +bad=types.ModuleType('phase4an_sealed_am_overlay');sys.modules[bad.__name__]=bad +try: + c.R._load(bad.__name__,p.parent/'wholebrain-plastic-interface-v1/overlay.py',E.SOURCE_PINS['am_overlay_source']) +except ValueError as e: assert 'Existing unverified module binding' in str(e) +else: raise AssertionError('poisoned helper accepted') +assert sys.modules[bad.__name__] is bad +print('poison-rejected') +""" + self.assertIn('poison-rejected', self.run_python(code)) + + def test_C_read_only_loader_after_ordinary_R_does_not_itself_load_overlay(self): + code = """ +import pathlib,sys +import runtime_light as R +p=pathlib.Path.cwd() +h=R._load('phase4an_sealed_am_overlay',p.parent/'wholebrain-plastic-interface-v1/overlay.py',R.AM_OVERLAY_SHA) +import runtime_continuation as C +assert 'phase4an_sealed_am_overlay' not in C.R._LOADED +assert sys.modules['phase4an_sealed_am_overlay'] is h +assert 'torch' not in sys.modules +print('checkpoint-reader-import-only-PASS') +""" + self.assertIn('checkpoint-reader-import-only-PASS', self.run_python(code)) + + def test_cli_bad_external_pin_exits_nonzero_no_outputs(self): + with tempfile.TemporaryDirectory() as tmp: + f = Fixture(Path(tmp), 'update') + result = subprocess.run([sys.executable, '-B', str(Path(w.__file__).resolve()), 'update', + '--job', str(f.job_file), '--job-sha256', 'f' * 64], capture_output=True, text=True, timeout=20) + self.assertNotEqual(result.returncode, 0) + self.assertIn('External job SHA mismatch', result.stderr) + self.assertFalse(f.output.exists()) + self.assertFalse(f.receipt.exists()) + + +if __name__ == '__main__': + unittest.main()