"""Pure normal-only outcome aggregation; NO disk, processes or new gates. outer_evidence MUST be supplied by the real terminal auditor after complete ledger/checksum and historical-integrity reconciliation, and an actual parent-observed successful invocation of the fixed normal_metrics.analyze. The caller must authenticate that process, source, consumed authority and saved report bytes. analysis_sha256 below binds canonical parsed JSON, NOT the raw pretty-printed file SHA (the caller must authenticate that separately). This function cannot establish any of those external facts from a Boolean or a digest. It checks their declared binding and the frozen report's internal structure/predicates only. Normal PASS is not candidate, usable learning, behavior, full legacy-H reproduction, or permission to run later assays. """ from __future__ import annotations import hashlib import json import math import numpy as np ANALYSIS_SOURCE_SHA256 = '82af63d8fadd5a6b67083f5e303ebe88674b76f9a2a8a8fff082b5c828483533' METRICS_SOURCE_SHA256 = '47319e21db3c8b030b1bafbf0703cd86fb588d23bb29436accbacdba73d08c3c' CANONICAL_SCOPE = "JSON sort_keys=True separators=(',',':') ensure_ascii=False allow_nan=False; UTF-8; no newline" GATE_SCHEMA = 'conditioning-learning-gates-v1' H_MAPPING = 'each-DAN-control-source-separately-both-pass-v1' MODES = ('physiological', 'matched_energy') DANS = ('PAM15', 'PPL103') CUES = ('4-ethylphenol', '3-hexanone') ARMS = {'paired': 'paired', 'd_zero': 'D=0', 'frozen': 'frozen', 'random_reward': 'random reward', 'unpaired': 'unpaired', 'cs_only': 'CS-only', 'dan_only': 'DAN-only'} CONTROLS = tuple(ARMS.values())[1:] POINTS = (0, 1, 2, 5, 10, 20, 50) POPS = ('MBON', 'legacyDownstream', 'nonMBONDownstream') PREDICATES = ('raw_signed_direction', 'beyond_each_actual_control', 'conditioned_final_g_exceeds_neutral', 'conditioned_count_or_defined_population_first_latency_change', 'all_MBON_AB_euclidean_gain_positive', 'legacyDownstream_AB_euclidean_gain_positive', 'nonMBONDownstream_AB_euclidean_gain_positive') SCOPE_FALSE = ('external_authority_verified', 'scientific_learning_pass_claimed', 'candidate_generated', 'execution_authorized', 'data_backed_claimed') OUTER_KEYS = ('registration_sha256', 'terminal_reconciliation_pass', 'historical_integrity_after_pass', 'analysis_sha256', 'actual_parent_exit_code', 'analyzed_source_sha256') class Incomplete(ValueError): pass def require(ok, message): if not ok: raise ValueError(message) def analysis_digest(report): """Canonical content digest; caller separately authenticates saved bytes.""" return hashlib.sha256(json.dumps(report, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False).encode('utf-8')).hexdigest() def is_sha(value): return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value) def finite(value): require(type(value) in (int, float) and math.isfinite(value), 'Missing/nonfinite scientific scalar') return value def exact_inventory(values, expected, label): require(len(values) == len(set(values)), 'Duplicate ' + label) actual = set(values) require(actual <= expected, 'Unexpected ' + label) if actual != expected: raise Incomplete('Missing ' + label) def scope(value): require(value['schema'] == GATE_SCHEMA and all(value[k] is False for k in SCOPE_FALSE) and value['frozen_metrics_source_sha256'] == METRICS_SOURCE_SHA256, 'Unsupported conditional-report authority claim/source') def compact_curve(row, blocks): """Check existing compact fields, not unretained raw-vector calculations.""" metric = row['metrics']; cue = blocks[row['block_id']]['conditioned_cue'] neutral = next(c for c in CUES if c != cue) raw = metric['rawB'] for phase in ('pre', 'post'): require(set(raw[phase]) == set(CUES), 'Curve cue coverage differs') for v in raw[phase].values(): finite(v) require(finite(metric['rawS']) == (raw['post'][cue] - raw['post'][neutral]) - (raw['pre'][cue] - raw['pre'][neutral]), 'Compact raw S/B relation differs') require(set(metric['separation']) == set(POPS) and set(metric['cues']) == set(CUES), 'Curve population/cue coverage differs') for value in metric['separation'].values(): require(finite(value['gain']) == finite(value['post']) - finite(value['pre']), 'Compact separation relation differs') for value in metric['cues'].values(): require(set(value['populations']) == set(POPS) | {'MBON01', 'MBON12'}, 'Missing compact population') for field in ('state_pre', 'state_post'): require(set(value[field]) == {'step', 'neuron_idx', 'counts', 'v', 'g', 'ref'} and all(is_sha(v) for v in value[field].values()), 'Missing compact state hashes') def _summarize(report, outer): if not isinstance(outer, dict) or any(k not in outer or outer[k] is None for k in OUTER_KEYS): raise Incomplete('Complete outer audit and actual analysis process evidence required') for key in ('terminal_reconciliation_pass', 'historical_integrity_after_pass'): require(type(outer[key]) is bool and outer[key], 'Outer integrity not satisfied: ' + key) require(type(outer['actual_parent_exit_code']) is int, 'Actual analysis parent exit must be an integer') if outer['actual_parent_exit_code'] != 0: raise Incomplete('Analysis process did not complete successfully') require(outer['analyzed_source_sha256'] == ANALYSIS_SOURCE_SHA256 and is_sha(outer['registration_sha256']) and outer['analysis_sha256'] == analysis_digest(report), 'Analysis source/content/registration authority differs') require(report['schema'] == 'normal-scientific-analysis-v1' and report['status'] == 'CONDITIONAL-PENDING-OUTER-INTEGRITY' and report['registration_sha256'] == outer['registration_sha256'] and is_sha(report['normal_plan_sha256']), 'Wrong normal report identity/status') require(all(report[k] is False for k in ('subsequent_assays_authorized', 'candidate_generated', 'learning_pass_claimed', 'scientific_learning_pass_claimed', 'data_backed_claimed', 'external_full_tree_anatomy_historical_integrity_verified_here')) and report['subsequent_assays_status'] == 'NOT-RUN', 'Report contains out-of-scope science/authorization claims') if report['analyzed_probe_calls'] != 3920 or report['curve_points'] != 1960: raise Incomplete('Complete 3,920 probes and 1,960 curve points required') require(type(report['analyzed_probe_calls']) is type(report['curve_points']) is int, 'Inventory counts must be integers') registry, authority, a = report['registry'], report['authority'], report['analysis'] require(registry['schema'] == 'conditioning-gate-registry-descriptor-v1' and registry['protocol_sha256'] == outer['registration_sha256'] and authority['schema'] == 'conditioning-data-authority-descriptor-v1' and authority['registry_sha256'] == analysis_digest(registry) and authority['population_binding_sha256'] == registry['population_binding_sha256'] and all(is_sha(authority[k]) for k in ('verification_receipt_sha256', 'execution_inventory_sha256', 'population_binding_sha256')), 'Registry/authority descriptor binding differs') blocks = registry['blocks'] expected_cells = {(m, d, c) for m in MODES for d in DANS for c in CUES} identities = [(b['mode'], b['dan_type'], b['conditioned_cue'], b['replicate']) for b in blocks.values()] exact_inventory(identities, {(*cell, r) for cell in expected_cells for r in range(1, 6)}, '40 registered replicate identities') require(all(type(b['replicate']) is int for b in blocks.values()), 'Replicate labels must be integers') curves = a['curves'] curve_keys = [(r['block_id'], r['arm'], r['checkpoint']) for r in curves] exact_inventory(curve_keys, {(b, arm, p) for b in blocks for arm in ARMS for p in POINTS}, 'complete curve inventory') require(all(type(r['checkpoint']) is int for r in curves), 'Checkpoint labels must be integers') for row in curves: compact_curve(row, blocks) terminal = {(r['block_id'], r['arm']): r['metrics'] for r in curves if r['checkpoint'] == 50} cells = a['cells']; cell_keys = [(c['mode'], c['dan_type'], c['conditioned_cue']) for c in cells] exact_inventory(cell_keys, expected_cells, 'eight normal cells') table, cell_lookup = [], {} for cell, identity in zip(cells, cell_keys): scope(cell) require(cell['assay'] == 'normal' and cell['registry_sha256'] == authority['registry_sha256'] and cell['external_data_authority_descriptor_sha256'] == analysis_digest(authority) and type(cell['required_majority']) is int and cell['required_majority'] == 3 and type(cell['training_replicates']) is int and cell['training_replicates'] == 5 and cell['acquisition_controls'] == list(CONTROLS) and cell['downstream_policy'] == 'unchanged acquisition requires both legacyDownstream and nonMBONDownstream positive gains', 'Cell rule/authority/control coverage differs') expected_ids = {key for key, b in blocks.items() if (b['mode'], b['dan_type'], b['conditioned_cue']) == identity} rows = cell['replicates'] exact_inventory([r['replicate_id'] for r in rows], expected_ids, 'same-cell independent replicates') for row in rows: block_id = row['replicate_id']; paired = terminal[(block_id, 'paired')] s = finite(paired['rawS']) controls = {legacy: finite(terminal[(block_id, name)]['rawS']) for name, legacy in ARMS.items() if name != 'paired'} beyond = {name: s > value if identity[1] == 'PAM15' else s < value for name, value in controls.items()} cue = identity[2]; neutral = next(c for c in CUES if c != cue) conditioned = paired['cues'][cue]['populations']['MBON']; other = paired['cues'][neutral]['populations']['MBON'] require(type(conditioned['spike_count_l1']) is int and conditioned['spike_count_l1'] >= 0 and type(conditioned['defined_first_latency_changed']) is bool, 'Functional count/latency schema differs') gains = {p: finite(paired['separation'][p]['gain']) for p in POPS} predicates = dict(zip(PREDICATES, ( s > 0 if identity[1] == 'PAM15' else s < 0, all(beyond.values()), finite(conditioned['final_delta_g_max_abs']) > finite(other['final_delta_g_max_abs']), conditioned['spike_count_l1'] > 0 or conditioned['defined_first_latency_changed'], gains['MBON'] > 0, gains['legacyDownstream'] > 0, gains['nonMBONDownstream'] > 0))) require(row['rawS'] == s and row['raw_control_S'] == controls and row['population_separation_gains'] == gains and row['beyond_controls'] == beyond and all(type(x) is bool for x in row['beyond_controls'].values()) and row['predicates'] == predicates and all(type(x) is bool for x in row['predicates'].values()) and row['conjunction_pass'] is all(predicates.values()), 'Replicate differs from frozen terminal conjunction') count = sum(r['conjunction_pass'] for r in rows); passed = count >= 3 require(type(cell['successful_same_replicate_conjunctions']) is int and cell['successful_same_replicate_conjunctions'] == count and cell['predicate_pass'] is passed and cell['status'] == ('CONDITIONAL-PREDICATES-PASS' if passed else 'CONDITIONAL-PREDICATES-FAIL'), 'Cell majority/result differs from same-replicate conjunctions') require(set(cell['legacy_control_MBON_gains']) == {'D=0', 'random reward', 'frozen'}, 'Missing H controls') for name, legacy in (('d_zero', 'D=0'), ('random_reward', 'random reward'), ('frozen', 'frozen')): require(cell['legacy_control_MBON_gains'][legacy] == {b: terminal[(b, name)]['separation']['MBON']['gain'] for b in expected_ids}, 'H controls differ from terminal source curves') table.append(dict(mode=identity[0], dan_type=identity[1], conditioned_cue=identity[2], assay='normal', status='CONDITIONAL-PASS' if passed else 'CONDITIONAL-FAIL', successful_replicates=count)) cell_lookup[identity] = cell cohort = a['cohort']; scope(cohort) normal_pass = all(c['predicate_pass'] for c in cells) require(cohort['status'] == 'COMPLETE' and cohort['required_assays'] == ['normal'] and cohort['complete_inventory'] is True and cohort['legacy_H_and_integrity_gates_included'] is False and cohort['all_functional_predicates_pass'] is normal_pass and cohort['table'] == sorted(table, key=lambda r: (r['mode'], r['dan_type'], r['conditioned_cue'], r['assay'])), 'Cohort summary differs') legacy = a['legacy_H']; scope(legacy) require(legacy['status'] == 'CONDITIONAL-AUDIT' and legacy['declared_mapping'] == H_MAPPING and legacy['original_H_data_layout_reproduced'] is False and legacy['new_stricter_control_source_mapping'] is True, 'H mapping changed') exact_inventory([(g['mode'], g['conditioned_cue']) for g in legacy['groups']], {(m, c) for m in MODES for c in CUES}, 'four separate H groups') for group in legacy['groups']: require(set(group['control_sources']) == set(DANS), 'Both H control sources required') for source, value in group['control_sources'].items(): means = value['protocol_means'] require(set(means) == {'appetitive-conditioning', 'aversive-conditioning', 'd-zero-control', 'random-reward-control', 'frozen-brain-control'}, 'H mean labels differ') for v in means.values(): finite(v) # Same five-element NumPy mean as frozen metrics.py; this is a # consistency check on the existing H mapping, not another gate. expected_means = {} for dan, label in (('PAM15', 'appetitive-conditioning'), ('PPL103', 'aversive-conditioning')): rows = cell_lookup[(group['mode'], dan, group['conditioned_cue'])]['replicates'] expected_means[label] = float(np.mean([r['population_separation_gains']['MBON'] for r in rows])) controls = cell_lookup[(group['mode'], source, group['conditioned_cue'])] for name, label in (('D=0', 'd-zero-control'), ('random reward', 'random-reward-control'), ('frozen', 'frozen-brain-control')): expected_means[label] = float(np.mean([controls['legacy_control_MBON_gains'][name][r['replicate_id']] for r in controls['replicates']])) require(means == expected_means, 'H means differ from exact branch/replicate control sources') floor = min(means['appetitive-conditioning'], means['aversive-conditioning']) ceiling = max(means[n] for n in ('d-zero-control', 'random-reward-control', 'frozen-brain-control')) require(value['schema'] == 'phase4an-metrics-draft-v1' and value['audit'] == 'legacy_H_mean_control_predicate_only' and value['source_reference'] == 'frozen phase4h.py:412-414' and value['AN_to_legacy_protocol_mapping_authority'] == 'caller; no automatic pooling' and value['condition_floor'] == floor and value['control_ceiling'] == ceiling and value['predicate_pass'] is (floor > ceiling) and value['full_old_H_gate_computed'] is False and value['scientific_learning_pass_claimed'] is False, 'Frozen strict H scalar predicate differs') require(group['both_pass'] is all(v['predicate_pass'] for v in group['control_sources'].values()), 'H group conjunction differs') h_pass = all(g['both_pass'] for g in legacy['groups']) require(legacy['both_pass'] is h_pass, 'H total conjunction differs') return dict(status='NORMAL-ACQUISITION-PASS' if normal_pass and h_pass else 'NORMAL-ACQUISITION-FAIL', normal_acquisition_pass=normal_pass and h_pass, all_eight_cell_conjunctions_pass=normal_pass, both_H_control_sources_all_groups_pass=h_pass, cell_table=cohort['table'], registration_sha256=outer['registration_sha256'], analysis_sha256=outer['analysis_sha256']) def summarize_normal_analysis(report, outer_evidence): """Classify only complete, bound normal reports; missing evidence is not FAIL.""" base = dict(schema='normal-outcome-summary-v1', learning_pass_claimed=False, usable_learning_pass=False, scientific_learning_pass_claimed=False, candidate_generated=False, subsequent_assays_authorized=False, subsequent_assays={'delayed': 'NOT-RUN', 'reversal': 'NOT-RUN', 'retention': 'NOT-RUN'}, outer_evidence_authenticated_here=False, raw_scientific_arrays_recomputed_here=False, canonical_scope=CANONICAL_SCOPE, scope='Registered normal acquisition only; external audit and process authentication are caller obligations') try: return dict(base, **_summarize(report, outer_evidence)) except (Incomplete, KeyError) as error: return dict(base, status='INCOMPLETE', normal_acquisition_pass=None, reason=str(error)) except (ValueError, TypeError, AttributeError, IndexError) as error: return dict(base, status='INTEGRITY-FAIL', normal_acquisition_pass=None, reason=str(error))