Digital Drosophila Archive
157 lines
11 KiB
Python
157 lines
11 KiB
Python
"""Recheck raw evidence and seal this explicit scheduling hypothesis, not learning."""
|
|
from context import ROOT, MODEL, EXP, AG, CFG, check, save, read, digest, integrity, budget, seal
|
|
|
|
|
|
def optional(path):
|
|
return read(path) if path.exists() else None
|
|
|
|
|
|
def main():
|
|
from compare_timing import compare, compare_gpu_repeats
|
|
from calibration import control_identity, summarize
|
|
check(); budget()
|
|
arms = ['delay' + str(delay) for delay in CFG['delays_steps']]
|
|
expected_cases = CFG['fixtures'] + CFG['calibration']['full_oracle_probes']
|
|
comparisons, repeats, controls, components = {}, {}, {}, {}
|
|
for arm in arms:
|
|
for name in expected_cases:
|
|
path = EXP / 'audit' / f'compare-{arm}-{name}.json'
|
|
if path.exists():
|
|
actual = compare(EXP / 'cases' / arm / name, EXP / 'gpu' / arm / (name + '-r0'), EXP / 'oracle' / arm / name, None)
|
|
assert actual == read(path), 'Changed raw oracle evidence: ' + arm + '/' + name
|
|
comparisons[arm + '/' + name] = actual['pass_all']
|
|
path = EXP / 'component-audit' / arm / (name + '.json')
|
|
if path.exists():
|
|
report = read(path)
|
|
assert set(report.get('input_sha256', {})) == {'state.npz', 'summary.json',
|
|
'apl-drive.float32.npy', 'apl-state.float32.npy', 'apl-inhibition.float32.npy'}
|
|
for filename, sha in report['input_sha256'].items():
|
|
assert digest(EXP / 'gpu' / arm / (name + '-r0') / filename) == sha
|
|
assert set(report.get('matrix_checks', {})) == {'input_matrix', 'output_matrix', 'input_raw', 'output_raw',
|
|
'kc_globals', 'apl_globals', 'pn_globals', 'apl_out_ids', 'input_scale', 'output_scale'}
|
|
assert set(report.get('ast_replay_bitwise_exact', {})) == {'drive', 'state', 'inhibition'}
|
|
assert set(report.get('float64_measurement', {})) == {'drive', 'state', 'inhibition'}
|
|
actual_pass = bool(not report.get('error') and all(report['matrix_checks'].values()) and
|
|
all(report['ast_replay_bitwise_exact'].values()) and report['theoretical_bound']['applicable'] and
|
|
all(metric['bound_violations'] == 0 for metric in report['float64_measurement'].values()))
|
|
assert report['pass_all'] == actual_pass
|
|
components[arm + '/' + name] = actual_pass
|
|
for name in CFG['fixtures']:
|
|
for repeat in range(1, CFG['fixture_repeats']):
|
|
path = EXP / 'audit' / f'repeat-{arm}-{name}-r{repeat}.json'
|
|
if path.exists():
|
|
actual = compare_gpu_repeats(EXP / 'gpu' / arm / (name + '-r0'), EXP / 'gpu' / arm / f'{name}-r{repeat}', None)
|
|
assert actual == read(path), 'Changed raw repeat evidence'
|
|
repeats[f'{arm}/{name}-r{repeat}'] = actual['pass_all']
|
|
for name in expected_cases:
|
|
path = EXP / 'audit' / f'control-{name}.json'
|
|
if path.exists():
|
|
actual = control_identity(EXP / 'gpu/delay0' / (name + '-r0'), AG / 'gpu' / (name + '-r0'), EXP / 'cases/delay0' / name)
|
|
assert actual == read(path), 'Changed delay0 control evidence'
|
|
controls[name] = actual['pass_all']
|
|
origin = optional(EXP / 'origin-audit/summary.json')
|
|
cal_records = read(EXP / 'protocol/calibration-cases.json')
|
|
origin_complete = False
|
|
if origin:
|
|
expected_origin = {row['name']: row for row in cal_records}
|
|
assert len(expected_origin) == len(cal_records) == 60
|
|
assert len(origin['trials']) == 60 and {row['case'] for row in origin['trials']} == set(expected_origin)
|
|
for row in origin['trials']:
|
|
original = expected_origin[row['case']]
|
|
assert row['repeat'] == 0 and row['ALLN_targets'] == 420 and row['pass_all']
|
|
assert set(row['exact']) == {'step', 'neuron_idx', 'counts', 'v', 'g', 'ref'} and all(row['exact'].values())
|
|
assert all(row[key] == original[key] for key in ('mode', 'odor_id', 'seed'))
|
|
origin_complete = True
|
|
for row in origin['trials']:
|
|
trial = EXP / 'origin-audit' / row['case']
|
|
assert digest(trial / 'summary.json') == row['summary_sha256']
|
|
for payload in read(trial / 'output-hashes.json'):
|
|
assert digest(trial / payload['path']) == payload['sha256']
|
|
for filename, sha in read(trial / 'summary.json')['input_sha256'].items():
|
|
assert digest(ROOT / filename) == sha
|
|
shift = optional(EXP / 'audit/shift-gate.json')
|
|
shift_complete = False
|
|
if shift:
|
|
expected_shift = {(name, delay) for name in CFG['fixtures'] + [row['name'] for row in cal_records] for delay in CFG['delays_steps']}
|
|
assert len(shift['records']) == len(expected_shift) == 136
|
|
assert {(row['case'], row['delay_steps']) for row in shift['records']} == expected_shift
|
|
assert all(row['pass_all'] for row in shift['records'])
|
|
shift_complete = True
|
|
from shift_audit import transport
|
|
import numpy as np
|
|
for row in shift['records']:
|
|
source = AG / 'gpu' / (row['case'] + '-r0') / 'apl-inhibition.float32.npy'
|
|
assert digest(source) == row['input_sha256']
|
|
_, _, actual = transport(np.load(source, mmap_mode='r', allow_pickle=False), row['delay_steps'])
|
|
assert all(row[key] == value for key, value in actual.items())
|
|
cal = optional(EXP / 'calibration/summary.json')
|
|
if cal is not None:
|
|
assert summarize(out_json=None) == cal, 'CAL scientific evidence changed'
|
|
expected = {arm + '/' + name for arm in arms for name in expected_cases}
|
|
expected_repeats = {f'{arm}/{name}-r{repeat}' for arm in arms for name in CFG['fixtures'] for repeat in range(1, CFG['fixture_repeats'])}
|
|
numerical = bool(origin_complete and origin['pass_all'] and shift_complete and shift['pass_all'] and
|
|
set(comparisons) == set(components) == expected and set(repeats) == expected_repeats and
|
|
set(controls) == set(expected_cases) and all(comparisons.values()) and all(components.values()) and
|
|
all(repeats.values()) and all(controls.values()))
|
|
gates = {name: optional(EXP / 'audit' / (name + '-gate.json')) for name in ('fixtures', 'probes', 'calibration')}
|
|
numerical = numerical and all(gates[name] and gates[name]['pass_all'] for name in ('fixtures', 'probes'))
|
|
cal_execution = bool(cal and cal['execution_gate_pass'] and gates['calibration'] and gates['calibration']['pass_all'])
|
|
treatment_pass = bool(cal and cal.get('arms', {}).get('delay18', {}).get('original_representation_gate_pass', False))
|
|
baseline_pass = bool(cal and cal.get('arms', {}).get('delay0', {}).get('original_representation_gate_pass', False))
|
|
frozen = read(EXP / 'regression/frozen-five.json')
|
|
after = integrity()
|
|
save(EXP / 'regression/old-integrity-after.json', after)
|
|
oldpass = bool(frozen['pass_all'] and all(row['pass'] for row in after.values()))
|
|
if not oldpass:
|
|
status = 'FAIL-FROZEN-INTEGRITY-OR-REGRESSION'
|
|
elif numerical and cal_execution:
|
|
status = 'PASS-FIXED-CAL-APL-TIMING-HYPOTHESIS' if treatment_pass else 'FAIL-FIXED-CAL-APL-TIMING-HYPOTHESIS'
|
|
elif (any(not passed for group in (comparisons, components, repeats, controls) for passed in group.values()) or
|
|
any(report is not None and not report['pass_all'] for report in (origin, shift, *gates.values())) or
|
|
cal is not None and not cal['execution_gate_pass']):
|
|
status = 'FAIL-NUMERICAL-OR-CONTROL-GATE'
|
|
else:
|
|
status = 'INCOMPLETE-NUMERICAL-OR-CAL-EXECUTION'
|
|
measurements = []
|
|
for path in sorted((EXP / 'gpu').glob('*/*/summary.json')):
|
|
summary = read(path)
|
|
measurements.append(dict(arm=path.parent.parent.name, run=path.parent.name,
|
|
**{key: summary[key] for key in ('case', 'total_spikes', 'accepted_requests', 'nodes', 'edges',
|
|
'diagnostic_simulation_seconds', 'peak_allocated_vram_bytes')}))
|
|
result = dict(status=status, numerical_gate_pass=bool(numerical), full_CAL_execution_gate_pass=cal_execution,
|
|
delay0_representation_pass=baseline_pass, delay18_representation_pass=treatment_pass,
|
|
origin_audit=origin, fixed_trace_shift=shift, gates=gates, comparisons=comparisons,
|
|
repeats=repeats, controls=controls, components=components, calibration=cal,
|
|
stopped=[dict(path=str(path.relative_to(EXP)), report=read(path)) for path in sorted((EXP / 'audit').glob('stopped-*.json'))],
|
|
old_frozen_integrity_and_five_reference_pass=oldpass, measurements=measurements,
|
|
unchanged_APL_parameters=CFG['apl'], explicit_hypothesis=CFG['hypothesis'],
|
|
delay_steps=CFG['delays_steps'], biological_delay_measured=False,
|
|
APL_nonKC_fast_edges_unreplaced=1212, shared_APL_arithmetic_oracle_limitation=True,
|
|
whole_brain_equal_inhibition_integral_claim=False, universal_equivalence_claim=False,
|
|
heldout_generalization_tested=False, learning_achieved=False, candidate_generated=False,
|
|
payload_bytes_before_manifest=budget(),
|
|
next_gate='Independent seeds and novel-odor representation; then credit specificity and functional learning' if numerical and cal_execution and treatment_pass
|
|
else 'Report the first failed gate for this fixed hypothesis; no tuning or conditioning promotion')
|
|
save(EXP / 'audit/final-status.json', result)
|
|
with (EXP / 'REPORT.md').open('x', encoding='utf-8') as stream:
|
|
stream.write('\n'.join(['# Phase 4AJ — fixed graded-APL timing branch', '', f'Status: `{status}`.', '',
|
|
f'Numerical gate: {numerical}. Complete both CAL arms: {cal_execution}.',
|
|
f'Original representation gate: delay0={baseline_pass}; delay18={treatment_pass}.',
|
|
'Only generated h application time changes: zero versus 18 steps. No brain gain, input, learning or mask change.',
|
|
'This is an engineering scheduling hypothesis, not measured MaleCNS APL delay. It delays PN- and KC-derived h together.',
|
|
'Finite stored generated samples are preserved in applied output or pending tail; the brain is not advanced to flush the tail.',
|
|
'Closed-loop generated h and accepted requests can differ between arms; equal total inhibition is not assumed.',
|
|
'Independent Brian LIF and candidate share legacy Torch APL arithmetic; component and separate delay-queue audits qualify this limit.',
|
|
'Original fast APL mask still leaves 1,212 non-KC edges unreplaced. No silent scope change.',
|
|
'The origin audit identifies timed real-edge antecedents, not unique or necessary biological causes.',
|
|
f'Frozen checksums and five reference trials: {oldpass}.',
|
|
'Costs include diagnostic synchronization, IPC, hashes and tracing; these are not production runtime benchmarks.',
|
|
'No heldout trajectories, conditioning, learned model, reward interface, game, decoder or candidate.',
|
|
'Next gate: ' + result['next_gate'], '']))
|
|
check(); seal(MODEL); seal(EXP)
|
|
print(status, flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|