diff --git a/models/male-cns-lif/conditioning-acceleration-v1/audit_integrity.py b/models/male-cns-lif/conditioning-acceleration-v1/audit_integrity.py new file mode 100644 index 0000000..df7219a --- /dev/null +++ b/models/male-cns-lif/conditioning-acceleration-v1/audit_integrity.py @@ -0,0 +1,49 @@ +"""Read-only rehash of the saved Phase4AR shutdown inventory and registered sources.""" +import argparse +from datetime import datetime, timezone +import hashlib +import json +from pathlib import Path +import subprocess +import time + +INVENTORY = Path('/home/qishizhe/phase4ar-user-shutdown-20260930T203113Z/saved-files.sha256.jsonl') +INVENTORY_SHA = 'd7b675b0d0d4647d8fe2fcf9616b9a599e5bc3c15565e604e0d66a046afb835a' +REG = Path('/data/digital-drosophila/experiments/phase4ar-normal-conditioning/protocol/preregistered.json') +REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740' + +def sha(path): + h = hashlib.sha256() + with Path(path).open('rb') as f: + for chunk in iter(lambda: f.read(8 << 20), b''): h.update(chunk) + return h.hexdigest() + +p = argparse.ArgumentParser() +p.add_argument('--output', type=Path, required=True) +a = p.parse_args() +assert not a.output.exists() and a.output.parent.is_dir() +start = time.perf_counter() +assert sha(INVENTORY) == INVENTORY_SHA and sha(REG) == REG_SHA +records = [json.loads(line) for line in INVENTORY.read_text().splitlines()] +differences = [] +for row in records: + path = Path(row['path']) + if not path.is_file() or path.stat().st_size != row['bytes'] or sha(path) != row['sha256']: + differences.append(str(path)) +reg = json.loads(REG.read_text()) +source_mismatches = [path for path, expected in reg['source_sha256'].items() if sha(path) != expected] +head = subprocess.check_output(['git', '-C', '/data/digital-drosophila', 'rev-parse', 'HEAD'], text=True).strip() +status = subprocess.check_output(['git', '-C', '/data/digital-drosophila', 'status', '--porcelain'], text=True) +summary = dict(status='PASS' if not differences and not source_mismatches and status == '' else 'FAIL', + created_utc=datetime.now(timezone.utc).isoformat(), files_checked=len(records), + bytes_checked=sum(r['bytes'] for r in records), inventory_sha256=INVENTORY_SHA, + registration_sha256=REG_SHA, registered_science_commit=reg['source_code_commit'], + sources_checked=len(reg['source_sha256']), file_mismatches=differences, + source_mismatches=source_mismatches, original_git_head=head, original_git_status=status, + wall_seconds=time.perf_counter()-start, + scope='Saved Phase4AR experiment/control files and registered source files; NOT a new Phase1-4AQ numerical regression', + training_epochs_cpu_verified_unchanged=1033, probes_completed_unchanged=138, + original_experiment_resumed=False) +with a.output.open('x') as f: json.dump(summary, f, indent=2) +print(json.dumps(summary), flush=True) +raise SystemExit(0 if summary['status'] == 'PASS' else 1) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py b/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py index c6be031..efa5662 100644 --- a/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py +++ b/models/male-cns-lif/conditioning-acceleration-v1/benchmark.py @@ -97,9 +97,16 @@ def load_fixtures(manifests): common_identity = identity assert common_identity == identity, 'Full KC/mask/base identity differs across fixtures' assert len(a['kc']) == 4064 and len(a['candidate_edge_id']) == 14232 - source_pins = read(npz.parent / 'reference-sources.json') - for info in source_pins.values(): - assert sha(info['path']) == info['sha256'], 'Reference source changed' + sources_path = npz.parent / 'reference-sources.json' + assert sha(sources_path) == m['files'][sources_path.relative_to(root).as_posix()]['sha256'] + source_pins = read(sources_path) + expected_sources = dict( + conditioning_update_source='96efff678cff910654eb08e86f1a17789e4e5589e340345f419c7027a7d547e2', + eligibility_bridge_source='174fb8d14619189e74b095e77a6db18b14c93046dc431c0b55710a5aa2dd2e61', + phase4j_source='abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e') + assert set(source_pins) == set(expected_sources) + for role, info in source_pins.items(): + assert info['sha256'] == expected_sources[role] and sha(info['path']) == info['sha256'], 'Reference source changed' # Extract constants from the authenticated original bridge, without import/exec. import ast bridge = source_pins['eligibility_bridge_source'] @@ -118,6 +125,33 @@ def load_fixtures(manifests): assert fixtures return fixtures, common_identity +def check_rejections(manifests, backend, rule): + reports = [] + for path in manifests: + for case in read(path).get('invalid_cases', []): + root = path.parent + npz, command = contained(root, case['npz']), contained(root, case['command']) + assert sha(npz) == case['npz_sha256'] and sha(command) == case['command_sha256'] + with np.load(npz, allow_pickle=False) as z: + a = {k: z[k].copy() for k in z.files} + c = read(command) + commands = [] if c['us_tick'] is None else [c['us_tick']] + stage = 'prepare' + try: + prepared = prepare_events(a['step'], a['neuron_idx'], a['kc'], c['steps'], commands, + n_neurons=c['n_neurons']) + stage = 'run' + backend.run(prepared, a['candidate_sources'], a['candidate_compartments'], + c['compartment'], a['delta_before'], a['base_candidate'], + d=c['d_command'], frozen=c['frozen'], **rule) + except ValueError as error: + assert stage == case['stage'], 'Rejected at a different validation boundary' + reports.append(dict(case_name=case['case_name'], rejected=True, + stage=stage, expected_exception=case['expected_exception'], message=str(error))) + else: + raise AssertionError('Candidate accepted invalid case: ' + case['case_name']) + return reports + def main(args): assert os.environ.get('CUDA_VISIBLE_DEVICES') == args.uuid, 'Use one explicit physical UUID per process' assert args.output.is_absolute() and not args.output.exists() @@ -193,11 +227,14 @@ def main(args): median_ms=1000 * report['summaries']['replay_wall_including_preparation_and_comparison_seconds']['median'])), flush=True) # Recheck original exported files at the end, not just at read time. load_fixtures(args.manifest) + rejections = check_rejections(args.manifest, backend, fixtures[0]['rule']) + save(args.output / 'invalid-input-rejections.json', rejections) result = dict(status='PASS-ELIGIBILITY-UPDATE-REPLAY-EQUIVALENCE', fixtures=len(reports), device=device, candidate_commit=commit, all_fixture_repeats_bitwise_exact=True, observed_cuda_pool_reserved_peak_bytes=peak_reserved, gpu_after=gpu_snapshot(args.uuid), queues_after=idle_queues(), wall_seconds=time.perf_counter()-started, original_experiment_resumed=False, brain_simulation_tested=False, + rejected_invalid_cases=len(rejections), full_training_speedup_measured=False, learning_success_claimed=False, coverage_limit='available prefix + fixed synthetic edge cases, not all registered conditions', files={p.name: sha(p) for p in sorted(args.output.iterdir()) if p.is_file()}) diff --git a/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py b/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py index 3bfee1b..1c7d997 100644 --- a/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py +++ b/models/male-cns-lif/conditioning-acceleration-v1/export_synthetic.py @@ -156,9 +156,113 @@ def fixed_cases(template): 'Synthetic generation changed an anatomical array: ' + name) original.require(not np.any((item['arrays']['delta_before'] == 0) & np.signbit(item['arrays']['delta_before'])), 'Synthetic delta has negative zero') + cases[4]['trace_command_steps'] = [0, 900, 1199] + cases[4]['tags'].append('multiple-US-snapshots') return cases +def supplemental_worker(directory): + """Fresh original CPU process for multiple snapshots or rejection evidence.""" + original.cpu_environment() + import numpy as np + directory = original.unlinked(directory) + sources = original.read(directory / 'reference-sources.json') + expected = dict(conditioning_update_source=original.U_SHA, + eligibility_bridge_source=original.BRIDGE_SHA, phase4j_source=original.J_SHA) + original.require(set(sources) == set(expected), 'Supplemental source roles differ') + for role, sha in expected.items(): + original.require(sources[role]['sha256'] == sha, 'Supplemental frozen source differs') + original.check(sources[role]['path'], sha) + if (directory / 'trace-command-steps.json').exists(): + arrays = original.load_npz(directory / 'inputs.npz') + command = original.read(directory / 'command.json') + commands = original.read(directory / 'trace-command-steps.json') + bridge = original.load_verified_source('_synthetic_original_bridge', + sources['eligibility_bridge_source']['path'], original.BRIDGE_SHA) + trace = bridge.trace_from_events(arrays['step'], arrays['neuron_idx'], arrays['kc'], + command['steps'], commands, phase4j_source=sources['phase4j_source']['path'], + backend='torch', n_neurons=command['n_neurons']) + sole = original.load_npz(directory / 'reference.npz') + original.require(original.array_evidence(trace.final) == original.array_evidence(sole['reference_final_trace']) + and original.array_evidence(trace.last_command_trace) + == original.array_evidence(sole['reference_sampled_trace']), + 'Multiple-US last sample or final differs from original single-US U reference') + original.save_npz(directory / 'multiple-reference.npz', dict(reference_snapshots=trace.snapshots, + reference_command_steps=trace.command_steps, reference_sampled_trace=trace.last_command_trace, + reference_final_trace=trace.final)) + import torch + original.require(not torch.cuda.is_initialized(), 'Supplemental CPU oracle initialized CUDA') + original.save(directory / 'multiple-reference.json', dict(complete=True, trace_audit=trace.audit, + source_pins=sources, numpy_version=np.__version__, torch_version=torch.__version__, + final_and_last_sample_match_original_U=True, brain_executed=False, weights_installed=False)) + return + cases = original.read(directory / 'invalid-selection.json') + update = original.load_verified_source('_synthetic_rejection_original_U', + sources['conditioning_update_source']['path'], original.U_SHA) + records = [] + for item in cases: + arrays = original.load_npz(directory / item['npz']) + command = original.read(directory / item['command']) + try: + update.update_epoch(**arrays, **command, + bridge_source=sources['eligibility_bridge_source']['path'], + phase4j_source=sources['phase4j_source']['path']) + except ValueError as error: + records.append(dict(case_name=item['case_name'], rejected=True, exception='ValueError', message=str(error))) + else: + raise ValueError('Original U unexpectedly accepted invalid case: ' + item['case_name']) + original.save(directory / 'original-rejections.json', dict(complete=True, cases=records, + source_pins=sources, brain_executed=False, weights_installed=False)) + + +def export_invalid(output, valid_case, sources): + import numpy as np + directory = output / 'invalid' + directory.mkdir() + original.save(directory / 'reference-sources.json', sources) + definitions = [('duplicate-events', 'prepare'), ('unsorted-events', 'prepare'), + ('out-of-range-tick', 'prepare'), ('negative-zero-delta', 'run'), + ('non-kc-candidate-source', 'run'), ('unsorted-kc', 'prepare')] + records = [] + for index, (name, stage) in enumerate(definitions): + arrays = {key: value.copy() for key, value in valid_case['arrays'].items()} + if name == 'duplicate-events': + arrays['step'] = np.array([0, 0], np.int32) + arrays['neuron_idx'] = np.full(2, arrays['kc'][0], np.int32) + elif name == 'unsorted-events': + arrays['step'] = np.array([1, 0], np.int32) + arrays['neuron_idx'] = np.full(2, arrays['kc'][0], np.int32) + elif name == 'out-of-range-tick': + arrays['step'] = np.array([STEPS], np.int32) + arrays['neuron_idx'] = np.array([arrays['kc'][0]], np.int32) + elif name == 'negative-zero-delta': + arrays['delta_before'][0] = np.float32(-0.) + elif name == 'non-kc-candidate-source': + arrays['candidate_sources'][0] = int(np.flatnonzero( + ~np.isin(np.arange(N_NEURONS, dtype=np.int32), arrays['kc']))[0]) + elif name == 'unsorted-kc': + arrays['kc'] = arrays['kc'][::-1].copy() + npz = f'{index:02d}-{name}.npz' + command = f'{index:02d}-{name}.json' + original.save_npz(directory / npz, arrays) + original.save(directory / command, valid_case['command']) + records.append(dict(case_name=name, npz=npz, command=command, stage=stage, + expected_exception='ValueError', npz_sha256=original.digest(directory / npz), + command_sha256=original.digest(directory / command))) + original.save(directory / 'invalid-selection.json', records) + original.run_cpu([sys.executable, '-B', str(Path(__file__).absolute()), '--supplemental-worker', str(directory)], + directory, 'original-rejection') + evidence = original.read(directory / 'original-rejections.json') + original.require(evidence['complete'] is True and len(evidence['cases']) == len(records), + 'Original rejection coverage incomplete') + for record, result in zip(records, evidence['cases']): + original.require(record['case_name'] == result['case_name'] and result['rejected'] is True, + 'Original rejection identity differs') + record.update(npz='invalid/' + record['npz'], command='invalid/' + record['command'], + original_oracle_rejection=result) + return records + + def export(template, output, metadata=None, manifest_path=None): original.cpu_environment() started = time.perf_counter() @@ -195,6 +299,11 @@ def export(template, output, metadata=None, manifest_path=None): for name in original.OUTPUT_ARRAYS: original.require(original.array_evidence(reference['reference_' + name]) == receipt['outputs'][name], 'Synthetic reference output/receipt differs: ' + name) + if 'trace_command_steps' in item: + original.save(directory / 'trace-command-steps.json', item['trace_command_steps']) + original.run_cpu([sys.executable, '-B', str(Path(__file__).absolute()), + '--supplemental-worker', str(directory)], directory, 'multiple-reference') + reference.update(original.load_npz(directory / 'multiple-reference.npz')) original.save_npz(directory / 'fixture.npz', {**item['arrays'], **reference}) record = dict(fixture_id=fixture_id, kind='synthetic', name=item['name'], tags=item['tags'], npz=f'{fixture_id}/fixture.npz', command=f'{fixture_id}/command.json', @@ -208,9 +317,13 @@ def export(template, output, metadata=None, manifest_path=None): reference_timing=timing, reference_subprocess_wall_seconds=seconds, input_arrays={name: original.array_evidence(value) for name, value in item['arrays'].items()}, output_arrays={name: original.array_evidence(value) for name, value in reference.items()}) + if 'trace_command_steps' in item: + record['snapshot_scope'] = 'Original Torch bridge multiple-US snapshots; last sample matches original U single final-US output' + record['command_steps'] = item['trace_command_steps'] original.save(directory / 'fixture.json', record) fixtures.append(record) print(original.canonical(dict(fixture=fixture_id, kind='synthetic', complete=True)), flush=True) + invalid_cases = export_invalid(output, cases[5], sources) for path, pin in pins.items(): original.check(path, pin) original.check(Path(__file__).absolute(), own) @@ -220,6 +333,7 @@ def export(template, output, metadata=None, manifest_path=None): exporter_source=dict(path=str(Path(__file__).absolute()), **own), original_helper_source=dict(path=str(Path(original.__file__).absolute()), **helper), created_utc=datetime.now(timezone.utc).isoformat(), fixtures=fixtures, fixture_count=len(fixtures), + invalid_cases=invalid_cases, invalid_case_count=len(invalid_cases), selection_sha256=original.digest(output / 'selection.json'), template_binding_sha256=original.digest(output / 'template-binding.json'), total_export_seconds=time.perf_counter() - started, original_files_modified=False, @@ -233,10 +347,16 @@ def export(template, output, metadata=None, manifest_path=None): if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('--template', type=Path, required=True) + parser.add_argument('--template', type=Path) parser.add_argument('--metadata', type=Path) parser.add_argument('--manifest', type=Path) - parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--output', type=Path) + parser.add_argument('--supplemental-worker', type=Path, help=argparse.SUPPRESS) args = parser.parse_args() - result = export(args.template, args.output, args.metadata, args.manifest) - print(original.canonical(dict(complete=True, output=str(args.output), fixtures=result['fixture_count'])), flush=True) + if args.supplemental_worker: + original.require(not any((args.template, args.output, args.metadata, args.manifest)), 'Do not combine worker/export modes') + supplemental_worker(args.supplemental_worker) + else: + original.require(args.template is not None and args.output is not None, '--template and --output are required') + result = export(args.template, args.output, args.metadata, args.manifest) + print(original.canonical(dict(complete=True, output=str(args.output), fixtures=result['fixture_count'])), flush=True)