Files
digital-drosophila/models/male-cns-lif/phase4n_finalize.py
T

263 lines
20 KiB
Python

#!/usr/bin/env python3
"""Phase 4N synthesis, independent checks, and final SHA-256 manifest."""
from __future__ import annotations
import argparse
import gc
import json
import resource
import shutil
import subprocess
from pathlib import Path
import numpy as np
import pandas as pd
import torch
from phase4h_audit import sha256, verify_checksum, write_json
from phase4i import phase4c_regression
from phase4k import dataset_checks, seal
from phase4m import frozen_checks, frozen_hashes
from phase4n import load_config
from phase4n_finish import curves, shared_decomposition
from phase4i import Validator
from phase4l import OdorSimulator
def contamination(exp, history):
event = pd.read_parquet(exp / "local-eligibility/event-level-updates.parquet")
if event.duplicated(["seed", "pairing", "edge_id"]).any():
raise RuntimeError("duplicate local eligibility row")
h = history[["train_seed", "pairing", "edge_id", "E_bounded_before", "E_bounded_after"]].rename(columns={"train_seed":"seed"})
x = event.merge(h,on=["seed","pairing","edge_id"],how="left",validate="one_to_one")
unmatched_runtime_rows=int(x.E_bounded_after.isna().sum())
nonmatching_trace_rows=int((~x.frozen_trace_match).sum())
excluded_B_rows=int(((~x.frozen_trace_match)&(x.identity_group=="B-specific")).sum())
x=x[x.frozen_trace_match & x.E_bounded_after.notna()].copy()
x["actual_E_depression"] = x.E_bounded_before-x.E_bounded_after
age = x.last_local_KC_spike_age_ms.to_numpy(np.float64)
x["temporal_credit_class"] = np.where(age <= 1.0,"coincident_<=1ms",np.where(age <= 20.0,"recent_1to20ms","long_trace_>20ms"))
x.to_parquet(exp / "b-contamination/updated-edges-with-temporal-source.parquet",index=False)
b=x[(x.identity_group=="B-specific")&(x.actual_E_depression>1e-7)]
total=float(b.actual_E_depression.sum())
fractions={str(k):float(g.actual_E_depression.sum()/total) if total else None for k,g in b.groupby("temporal_credit_class")}
event_fraction={str(k):int(len(g)) for k,g in b.groupby("temporal_credit_class")}
relative_short=np.divide(b.eligibility_short_shadow.to_numpy(np.float64),b.eligibility_v2.to_numpy(np.float64),
out=np.zeros(len(b),np.float64),where=b.eligibility_v2.to_numpy(np.float64)>0)
weighted_short=float(np.dot(b.actual_E_depression.to_numpy(np.float64),relative_short)/total) if total else None
report={"B_specific_update_rows":int(len(b)),"B_specific_unique_edges":int(b.edge_id.nunique()),
"B_specific_unique_source_KC":int(b.source_neuron_idx.nunique()),
"first_B_specific_update_pairing_by_seed":{str(seed):int(frame.pairing.min()) for seed,frame in b.groupby("seed")},
"B_specific_E_depression_sum_with_exact_replay":total,"temporal_credit_weighted_fraction":fractions,"temporal_credit_event_counts":event_fraction,
"excluded_nonmatching_trace_rows":nonmatching_trace_rows,"excluded_nonmatching_B_rows":excluded_B_rows,"unmatched_runtime_rows":unmatched_runtime_rows,
"external_matched_US_gate_fraction":1.0,"endogenous_DAN_as_gate_fraction_in_frozen_model":0.0,
"causal_route_fractions":{"direct_A_driven":None,"recurrent_KC":None,"MBON_DAN_feedback":None,"olfactory_feedback":None,"unresolved":1.0},
"reason":"Whole-brain source spike timing plus a positive anatomical edge is not proof of membrane-spike causation. Frozen Phase 4M history has no causal predecessor trace. Do not assign invented path percentages.",
"shadow_short_tau_20ms_excluded_B_update_fraction":float(b.loc[b.eligibility_short_shadow<.01,"actual_E_depression"].sum()/total) if total else None,
"shadow_short_tau_20ms_weighted_eligibility_ratio_to_frozen":weighted_short,
"event_coincidence_1ms_included_B_update_fraction":float(b.loc[b.event_coincidence_shadow,"actual_E_depression"].sum()/total) if total else None}
report["first_activation_only_included_B_update_fraction"]=(float(b.loc[b.first_activation_only_shadow,"actual_E_depression"].sum()/total) if total else None)
write_json(exp / "b-contamination/summary.json",report)
return report
def dan_summary(exp, identity, ncfg, cfg=None):
gates=pd.read_parquet(exp / "dan-source/event-level-gates.parquet")
trail=pd.read_parquet(exp / "endogenous-dan/shadow-trajectory.parquet")
source=gates.groupby("source").agg(gates=("source","size"),E_depression_sum=("E_change_sum","sum")).reset_index()
external_only=trail[trail.pairing==100].external_matched_E_reduction_sum.mean()
all_dan=trail[trail.pairing==100].all_DAN_shadow_E_reduction_sum.mean()
report={"source_breakdown":source.to_dict("records"),"trial_count":len(trail),
"matched_endogenous_DAN_spikes_total":int(trail.matched_endogenous_DAN_spikes.sum()),
"external_only_E_depression_sum_mean_100":float(external_only),"all_DAN_shadow_E_depression_sum_mean_100":float(all_dan),
"additional_all_DAN_shadow_E_depression_sum_mean_100":float(all_dan-external_only),
"frozen_Phase4M_endogenous_DAN_teaching_gate_count":0,
"all_DAN_shadow_interpretation":"counterfactual open-loop teaching-gate interpretation; not frozen model and not measured dopamine concentration"}
if cfg is not None:
mapping=pd.read_parquet(Path(cfg["paths"]["phase4h_root"])/"mappings/dan-to-mbon-confirmed.parquet")
mapping=mapping[["source_neuron_idx","source_annotation_type","source_compartment_verbatim"]].drop_duplicates()
matched_set=set(map(int,mapping.loc[mapping.source_compartment_verbatim.astype(str)==str(ncfg["matched_compartment_verbatim"]),"source_neuron_idx"]))
mapping=mapping[["source_neuron_idx","source_annotation_type"]].drop_duplicates(subset=["source_neuron_idx"])
events_file=exp/"dan-source/endogenous-DAN-spike-history.parquet"
if events_file.exists():
events=pd.read_parquet(events_file)
events=events.merge(mapping,left_on="neuron_idx",right_on="source_neuron_idx",how="left")
typ=events.source_annotation_type.fillna("unknown").astype(str)
events["cluster"] = np.where(typ.str.startswith("PAM"),"PAM",np.where(typ.str.startswith("PPL1"),"PPL1","other/unknown"))
report["endogenous_spike_cluster_counts"]={str(k):int(v) for k,v in events.cluster.value_counts().items()}
report["confirmed_type_counts"]={str(k):int(v) for k,v in typ.value_counts().items()}
report["matched_compartment_spikes"] = int(events.neuron_idx.isin(matched_set).sum())
events.to_parquet(exp/"dan-source/endogenous-DAN-with-type.parquet",index=False)
write_json(exp / "dan-source/summary.json",report)
return report
def shared_summary(exp):
p=exp / "shared-edge/isolated-decomposition.parquet"
if not p.exists(): return {"status":"NOT-RUN-NO-PEAK"}
table=pd.read_parquet(p)
agg=table.groupby(["decomposition","odor"]).agg(spike_L1=("MBON_spike_L1","mean"),transmission_reduction=("transmitted_reduction_percent","mean"),
latency_delta=("first_latency_delta_ms","mean")).reset_index()
write_json(exp / "shared-edge/summary.json",{"scope":"isolated replay only; existing edges never deleted", "group_means":agg.to_dict("records")})
return agg.to_dict("records")
def separation(exp, peak):
whole=pd.read_parquet(exp / "whole-brain/checkpoint-responses.parquet")
iso=pd.read_parquet(exp / "isolated/response-by-checkpoint.parquet")
rows=[]
for pair in sorted(whole.pairings.unique()):
w=whole[whole.pairings==pair].pivot(index="train_seed",columns="odor",values="MBON_spike_L1")
i=iso[iso.pairings==pair]
whole_sep=float((w["A"]-w["B"]).mean())
iso_sep=float(i.functional_separation_spike_L1.mean())
rows.append({"pairings":int(pair),"isolated_A_minus_B_spike_L1":iso_sep,"wholebrain_A_minus_B_spike_L1":whole_sep,
"wholebrain_specificity_penalty":iso_sep-whole_sep,
"wholebrain_A_specific_seeds":int(((w["A"]>w["B"])&(w["A"]>0)).sum())})
frame=pd.DataFrame(rows);frame.to_parquet(exp / "whole-brain/specificity-penalty.parquet",index=False)
return {"trajectory":rows,"at_peak":next((r for r in rows if r["pairings"]==peak),None)}
def control_summary(exp):
original=json.loads((exp / "controls/summary.json").read_text())
independent=json.loads((exp / "controls/independent-summary.json").read_text())
wholebrain=json.loads((exp / "controls/wholebrain-summary.json").read_text())
original.update({"B_matched_US":"RUN-INDEPENDENT-100-PAIRING-TRAJECTORY",
"random_US_timing":"RUN-INDEPENDENT-100-PAIRING-TRAJECTORY",
"unrelated_compartment_US":"RUN-INDEPENDENT-100-PAIRING-TRAJECTORY",
"all_DAN_gate":"RUN-OPEN-LOOP-SHADOW-100-PAIRING-TRAJECTORY",
"full_control_matrix_pass":False,"independent_results":"controls/independent-summary.json",
"all_DAN_closed_loop":False,"independent_wholebrain":"controls/wholebrain-summary.json"})
write_json(exp / "controls/summary.json",original)
return original, independent, wholebrain
def integrity(exp, ncfg, mcfg, cfg, basepath):
gc.collect();torch.cuda.empty_cache()
previous=frozen_checks(cfg)
mroot=Path(mcfg["experiment_root"]);mmodel=Path(mcfg["model_root"])
phase4m={"experiment":verify_checksum(mroot,mroot/"manifests/checksums.sha256")["passed"],
"model":verify_checksum(mmodel,mmodel/"checksums.sha256")["passed"]}
dataset=dataset_checks(cfg)
before=json.loads((exp / "audit/frozen-integrity.json").read_text())
# A second replay detects intermittent GPU/event divergence; it never rewrites frozen references.
replay=phase4c_regression(cfg,basepath,exp)
phase4m_summary=json.loads((mroot/"audit/final-summary.json").read_text())
reference_hashes=phase4m_summary["integrity"]["after"]
static={"previous_phases":previous["all_passed"],"phase4m":all(phase4m.values()),"phase1_to_3b":dataset["all_passed"],
"frozen_hashes_unchanged":frozen_hashes(cfg)==reference_hashes}
report={"static_checksums_all_passed":bool(all(static.values())),"static":static,
"phase4c_reference_replay_first":before["phase4c_reference_regression"],"phase4c_reference_replay_second":replay,
"dynamic_regression_all_passed":bool(before["phase4c_reference_regression"]["all_passed"] and replay["all_passed"]),
"all_passed":bool(all(static.values()) and before["phase4c_reference_regression"]["all_passed"] and replay["all_passed"])}
write_json(exp / "audit/frozen-integrity-final.json",report)
return report
def main():
ap=argparse.ArgumentParser();ap.add_argument("--config",type=Path,required=True);args=ap.parse_args()
ncfg,mcfg,cfg,basepath=load_config(args.config);exp=Path(ncfg["experiment_root"])
identity=pd.read_parquet(exp / "edge-identity/frozen-edge-identity.parquet")
response_curve=pd.read_parquet(exp/"specificity-curve/memory-specificity.parquet")
eligible=response_curve[response_curve.A_specific_functional_seeds>=3]
corrected_peak=None
if not eligible.empty:
best=float(eligible.mean_spike_L1_separation.max())
eligible=eligible[np.isclose(eligible.mean_spike_L1_separation,best,atol=1e-9,rtol=0)]
latency_best=float(eligible.mean_latency_separation_ms.max())
eligible=eligible[eligible.mean_latency_separation_ms>=latency_best-float(cfg["model"]["dt_ms"])/2]
corrected_peak=int(eligible.pairings.min())
old_peak=json.loads((exp/"specificity-curve/peak.json").read_text())["pairings"]
if corrected_peak!=old_peak:
write_json(exp/"audit/peak-numerical-tie-correction.json",{"original_checkpoint":old_peak,"corrected_checkpoint":corrected_peak,
"reason":"50 and 60 have the same five-seed spike-L1 separation; latency difference is sub-femtosecond float summation noise, far below 0.1 ms simulation dt. Earliest-tie rule selects 50.",
"not_a_learning_parameter_change":True,"identity_and_E_trajectories_unchanged":True})
write_json(exp/"specificity-curve/peak.json",{"status":"SPECIFICITY-PEAK-IDENTIFIED","pairings":corrected_peak,
"selection_rule":ncfg["peak_rule"],"numerical_tie_correction_record":"audit/peak-numerical-tie-correction.json"})
curves(ncfg,exp,pd.read_parquet(exp/"specificity-curve/CSI.parquet"),response_curve,identity)
validator=Validator(cfg,basepath,exp);sim=OdorSimulator(validator,cfg,exp)
shared_decomposition(ncfg,exp,identity,sim,validator,corrected_peak)
del sim,validator
stopping=json.loads((exp/"stopping-analysis/summary.json").read_text())
stopping["functional_peak_checkpoint"]=corrected_peak
write_json(exp/"stopping-analysis/summary.json",stopping)
# Complete the explicitly predeclared t=0 checkpoint without rerunning any brain trial.
groups0=pd.read_parquet(exp/"acquisition-trajectory/group-plasticity.parquet")
csi0=pd.read_parquet(exp/"specificity-curve/CSI.parquet")
if 0 not in set(groups0.pairings):
z=[];cz=[]
for seed in ncfg["train_seeds"]:
one=np.ones(len(identity),np.float32)
np.savez_compressed(exp/"acquisition-trajectory"/f"E-seed-{seed}-pairings-0.npz",E_bounded=one,E_unbounded=one,edge_id=identity.edge_id.to_numpy(np.int64))
for mechanism in ("bounded","unbounded"):
for group_name,count in identity.identity_group.value_counts().items():
z.append({"seed":seed,"pairings":0,"mechanism":mechanism,"group":group_name,"total_edges":int(count),
"updated_edges":0,"mean_delta_E":0.0,"median_delta_E":0.0,"weighted_transmission_reduction":0.0,"E_change_total":0.0})
cz.append({"seed":seed,"pairings":0,"mechanism":mechanism,"A-specific":0.0,"B-specific":0.0,"shared":0.0,"inactive":0.0,
"CSI":0.0,"A_to_shared_ratio":np.nan,"A_to_B_ratio":np.nan})
pd.concat([groups0,pd.DataFrame(z)],ignore_index=True).sort_values(["seed","pairings","mechanism","group"]).to_parquet(exp/"acquisition-trajectory/group-plasticity.parquet",index=False)
pd.concat([csi0,pd.DataFrame(cz)],ignore_index=True).sort_values(["seed","pairings","mechanism"]).to_parquet(exp/"specificity-curve/CSI.parquet",index=False)
history=pd.read_parquet(Path(mcfg["experiment_root"])/"efficacy-overlay/update-history.parquet")
peak=json.loads((exp/"specificity-curve/peak.json").read_text())["pairings"]
onset=json.loads((exp/"specificity-curve/generalization-onset.json").read_text())["pairings"]
contam=contamination(exp,history);dan=dan_summary(exp,identity,ncfg,cfg);shared=shared_summary(exp)
timed_inputs=json.loads((exp/"whole-brain/B-specific-timed-input-summary.json").read_text())
sep=separation(exp,peak);controls,independent,control_whole=control_summary(exp)
gate_specificity=json.loads((exp/"endogenous-dan/gate-specificity.json").read_text())
perf=json.loads((exp / "performance/summary.json").read_text())
audit_perf=json.loads((exp / "performance/event-audit.json").read_text())
ctrl_perf={"training_calls":independent["training_calls"],"training_wall_seconds":independent["training_wall_seconds"],
"mean_call_seconds":independent["mean_call_seconds"],"wholebrain_wall_seconds":control_whole["total_brain_wall_seconds"]}
perf.update({"event_audit":audit_perf,"controls":ctrl_perf,"peak_RAM_KiB_all_stages":max(perf["peak_RAM_KiB"],audit_perf["peak_RAM_KiB"],independent["peak_RAM_KiB"]),
"shadow_trace_overhead":"included in event-audit call time; no paired unaudited timing isolation"})
write_json(exp / "performance/summary.json",perf)
integ=integrity(exp,ncfg,mcfg,cfg,basepath)
event_reproduction=json.loads((exp/"regression/event-trace-reproduction.json").read_text())
csi=pd.read_parquet(exp / "specificity-curve/CSI.parquet")
group=pd.read_parquet(exp / "acquisition-trajectory/group-plasticity.parquet")
iso=pd.read_parquet(exp / "specificity-curve/memory-specificity.parquet")
long_fraction=contam["temporal_credit_weighted_fraction"].get("long_trace_>20ms")
shared_b=next((r["spike_L1"] for r in shared if r["decomposition"]=="A+shared" and r["odor"]=="B"),None) if isinstance(shared,list) else None
a_only_b=next((r["spike_L1"] for r in shared if r["decomposition"]=="A-only" and r["odor"]=="B"),None) if isinstance(shared,list) else None
mechanism_tags={"SPECIFICITY-PEAK-IDENTIFIED":peak is not None,
"GENERALIZATION-ONSET-IDENTIFIED":onset is not None,
"ENDOGENOUS-DAN-CREDIT-CONTAMINATION":False,
"RECURRENT-KC-CREDIT-CONTAMINATION":None,
"LONG-TRACE-CREDIT-CONTAMINATION":bool(long_fraction is not None and long_fraction>0),
"SHARED-EDGE-GENERALIZATION-DOMINANT":bool(shared_b is not None and a_only_b is not None and shared_b>a_only_b),
"NO-SINGLE-DOMINANT-CONTAMINATION-SOURCE":True}
audit_complete=bool(event_reproduction["all_passed"] and integ["all_passed"] and controls["full_control_matrix_pass"])
summary={"phase":"4N","status":"MECHANISM-AUDIT-COMPLETE-NO-CANDIDATE" if audit_complete else "PARTIAL-MECHANISM-AUDIT-NO-CANDIDATE",
"edge_groups":identity.identity_group.value_counts().to_dict(),"specificity_peak_pairings":peak,
"generalization_onset_pairings_by_predeclared_CSI_rule":onset,
"functional_collapse_first_checkpoint":next((int(r.pairings) for _,r in iso.iterrows() if r.pairings>peak and r.A_specific_functional_seeds<3),None) if peak is not None else None,
"group_curves":group[group.mechanism=="bounded"].groupby(["pairings","group"]).mean(numeric_only=True).reset_index()[["pairings","group","mean_delta_E","weighted_transmission_reduction"]].to_dict("records"),
"isolated_curve":iso.to_dict("records"),"CSI_curve":csi[csi.mechanism=="bounded"].groupby("pairings").CSI.mean().reset_index().to_dict("records"),
"B_specific_contamination":contam,"B_specific_timed_input_candidates":timed_inputs,"DAN":dan,"gate_specificity":gate_specificity,"shared_edge_isolated":shared,"wholebrain_specificity":sep,
"controls":controls,"control_wholebrain":control_whole,"event_trace_reproduction":event_reproduction,"mechanism_tags":mechanism_tags,"candidate_generated":False,"candidate_gate":{"external_vs_endogenous_distinguished":True,
"causal_B_contamination_route_identified":False,"isolated_and_wholebrain_majority_improvement_demonstrated":False},
"frozen_integrity":integ,"limitations":["Temporal predecessor class unresolved; no causal path attribution percentage claimed",
"All-DAN efficacy is an open-loop shadow, not a closed-loop retrained brain",
"Phase 4C mechanosensory frozen regression showed at least one divergent run; static SHA-256 passes but dynamic integrity cannot be called PASS"]}
write_json(exp / "audit/final-summary.json",summary)
model=Path(ncfg["model_root"]);model.mkdir(parents=True,exist_ok=True)
for name in ("phase4n.py","phase4n_finish.py","phase4n_events.py","phase4n_controls.py","phase4n_control_whole.py","phase4n_alldan.py","phase4n_paths.py","phase4n_preflight.py","phase4n_finalize.py","phase4n.yaml"):
source=Path(__file__).with_name(name)
shutil.copy2(source,model/name)
shutil.copy2(source,exp/"audit"/name)
write_json(model/"VERSION.json",{"phase":"4N","status":summary["status"],"candidate_generated":False,
"identity_sha256":sha256(exp/"edge-identity/frozen-edge-identity.parquet"),"config_sha256":sha256(args.config),
"frozen_static_integrity":integ["static_checksums_all_passed"],"frozen_dynamic_regression":integ["dynamic_regression_all_passed"]})
seal(model,model/"MANIFEST.json",model/"checksums.sha256")
seal(exp,exp/"manifests/manifest.json",exp/"manifests/checksums.sha256")
self_check={}
for root,relative in ((model,"checksums.sha256"),(exp,"manifests/checksums.sha256")):
result=subprocess.run(["sha256sum","-c",relative],cwd=root,stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,text=True)
self_check[str(root)]=result.returncode==0
if not all(self_check.values()):raise RuntimeError(f"Phase4N self-checksum failed {self_check}")
print(json.dumps({"status":summary["status"],"peak":peak,"onset":onset,"B_contamination":contam,
"DAN":dan,"wholebrain_penalty":sep["at_peak"],"integrity":integ,"self_checks":self_check}),flush=True)
if __name__=="__main__":main()