"""One explicitly exploratory, fixed-candidate audit of unweighted top-k rejection. Primary protocol remains unchanged/negative. A new confirmation uses untouched fresh texts; no further candidate selection, threshold tuning, weights or R fitting. """ import argparse import hashlib import json from pathlib import Path import resource import time import traceback import numpy as np from tokenizers import Tokenizer from functional.run import write_json,save_csv from .core import InterventionLM,distribution_metrics,output_metrics from .run import head_floor,rows_for CANDIDATE={'id':'conditional_qk_head_3_23_7_6','kind':'qk_head','i':3,'j':23,'a':7,'b':6,'perm':None} PROTOCOL={'classification':'post-screen exploratory metric audit; not a primary-protocol positive', 'candidate':CANDIDATE,'conditioning_input_layer':23,'donor_projection_layer':3,'target_head':6,'donor_head':7, 'selection_basis':'one minimum worst-text JS among the already measured source-conditioned pairs; no raw readout/fresh data used', 'js_max':.01,'js_row_p95_max':.05,'mean_total_variation_max':.10,'row_total_variation_p95_max':.20, 'target_head_output_relative_l2_max':.20,'target_head_output_cosine_min':.98, 'primary_query_positions':list(range(7,32)),'top5_is_diagnostic_only':True, 'active_head_required':True,'no_candidate_fallback':True,'fresh_text_file':'crossqkv/fresh_samples.json', 'same_input_network_direction':'layer23 head6 uses QK probability of layer3 head7 recomputed on H23', 'actual_cache_separate':'layer3 probability on H3 reused at23; different-input diagnostic, not function-equivalence evidence', 'quality_gate':{'mean_nll_delta_max':.01,'per_text_p95_nll_delta_max':.05},'training':False,'R_fitting':False,'cost_required':False} def tv_metrics(a,b): tv=.5*np.sum(np.abs(a.astype(np.float64)-b.astype(np.float64)),axis=-1)[:,:,7:] return {'total_variation':tv.mean((1,2)),'total_variation_p95':np.quantile(tv.reshape(tv.shape[0],-1),.95,axis=1)} def evaluate(lm,h,samples,split,donor_h=None): target=lm.parts(h,23);donor=lm.parts(h if donor_h is None else donor_h,3) a,b=target['p'][:,6:7],donor['p'][:,7:8] pm={**distribution_metrics(a,b),**tv_metrics(a,b)} changed=lm.head_transplant_output(target,donor,23,6,7) om=output_metrics(target['heads'][:,7:,6],changed[:,7:]);floor=head_floor(target) uniform=np.zeros_like(a) for t in range(a.shape[-1]):uniform[:,:,t,:t+1]=1/(t+1) null=distribution_metrics(a,uniform) rows=rows_for({**pm,**{f'output_{k}':v for k,v in om.items()},'target_uniform_js':null['js']},samples,{'split':split,'mode':'same_H23' if donor_h is None else 'actual_H3_to_H23'}) for n,r in enumerate(rows):r['active']=bool(om['rms_a'][n]>=floor[n] and om['rms_b'][n]>=floor[n]) passed=all(r['js']<=.01 and r['js_p95']<=.05 and r['total_variation']<=.10 and r['total_variation_p95']<=.20 and max(r['output_error_a'],r['output_error_b'])<=.20 and r['output_cosine']>=.98 and r['active'] for r in rows) return rows,passed def main(): p=argparse.ArgumentParser();p.add_argument('--source',type=Path,required=True);p.add_argument('--root',type=Path,required=True);a=p.parse_args();source=a.source.resolve();root=a.root.resolve() assert not (root/'conditional-protocol.json').exists(),'Fresh conditional audit protocol required' write_json(root/'conditional-protocol.json',{**PROTOCOL,'registered_unix':time.time(),'fresh_text_sha256':hashlib.sha256(Path(__file__).with_name('fresh_samples.json').read_bytes()).hexdigest()}) prior=json.loads((root/'status.json').read_text());assert prior['state']=='completed' and not prior['selected'] reserve=prior['elapsed_seconds']+60 remaining=int(7200-reserve-180) if remaining<120:raise TimeoutError('Insufficient shared two-hour budget') started=time.monotonic();status={'state':'running','phase':'calibration_readout','primary_protocol_unchanged_and_negative':True} rows=[] try: lm=InterventionLM(source/'weights',json.loads((source/'model_config.json').read_text()),seconds=remaining) old=json.loads((source/'samples-tokenized.json').read_text())['samples'];data=np.load(source/'baseline-activations.npz') for split in ('calibration','heldout'): index=np.array([i for i,s in enumerate(old) if s['split']==split]);samples=[old[i] for i in index] rr,passed=evaluate(lm,data['attn_23'][index],samples,split) rows.extend(rr);save_csv(root/'conditional-function-metrics.csv',rows) status[split+'_pass']=passed write_json(root/'conditional-status.json',status) if not passed: status['stop_reason']=f'{split} probability-mass/readout/activity screen failed';break else: fresh=json.loads(Path(__file__).with_name('fresh_samples.json').read_text())['samples'];tokenizer=Tokenizer.from_file(str(source/'tokenizer.json'));ids=[] for s in fresh: tokens=tokenizer.encode(s['text'],add_special_tokens=False).ids;assert len(tokens)>=32 s.update(token_ids=tokens[:32],full_token_count=len(tokens),text_sha256=hashlib.sha256(s['text'].encode()).hexdigest());ids.append(tokens[:32]) ids=np.array(ids,dtype=np.int64);write_json(root/'conditional-fresh-samples.json',{'samples':fresh,'candidate_already_fixed':True}) status['phase']='fresh_baseline';write_json(root/'conditional-status.json',status) h,inputs=lm.forward(ids,capture=[('attn',3),('attn',23)]) np.savez(root/'cache'/'conditional-fresh-inputs.npz',**inputs,final=h) base=lm.nll(h,ids);np.save(root/'conditional-fresh-baseline-token-nll.npy',base) rr,passed=evaluate(lm,inputs['attn_23'],fresh,'fresh_test');rows.extend(rr);save_csv(root/'conditional-function-metrics.csv',rows) status['fresh_function_pass']=passed if passed: status['phase']='one_direction_same_input_network';write_json(root/'conditional-status.json',status) lm.configure(CANDIDATE,reverse=True,cached=False);changed,_=lm.forward(ids);nll=lm.nll(changed,ids);np.save(root/'conditional-same-input-network-token-nll.npy',nll) delta=nll.mean(1)-base.mean(1) network={'mode':'layer23_head6_uses_layer3_head7_on_H23','sequence_nll_deltas':delta.tolist(),'mean_nll_delta':float(delta.mean()),'sequence_p95_nll_delta':float(np.quantile(delta,.95)), 'quality_pass':bool(delta.mean()<=.01 and np.quantile(delta,.95)<=.05),'one_direction_only':True} write_json(root/'conditional-network.json',network);status['same_input_network_quality_pass']=network['quality_pass'] # Check the actual-cache probability/readout separately, then run its network experiment. cached_rows,cached_function_pass=evaluate(lm,inputs['attn_23'],fresh,'fresh_test_actual_cache',donor_h=inputs['attn_3']) save_csv(root/'conditional-actual-cache-metrics.csv',cached_rows) status['actual_cache_function_screen_pass']=cached_function_pass lm.configure(CANDIDATE,reverse=True,cached=True);changed,_=lm.forward(ids);nll=lm.nll(changed,ids);np.save(root/'conditional-actual-cache-network-token-nll.npy',nll) delta=nll.mean(1)-base.mean(1) cache={'mode':'actual_layer3_H3_probability_reused_at23_H23','sequence_nll_deltas':delta.tolist(),'mean_nll_delta':float(delta.mean()),'sequence_p95_nll_delta':float(np.quantile(delta,.95)), 'quality_pass':bool(delta.mean()<=.01 and np.quantile(delta,.95)<=.05),'different_hidden_inputs':True} write_json(root/'conditional-actual-cache-network.json',cache);status['actual_cache_network_quality_pass']=cache['quality_pass'] status.update(state='completed',phase='done',conditional_function_positive=bool(status.get('fresh_function_pass')), position_pattern_classification='near_uniform' if rows and max(r['target_uniform_js'] for r in rows)<=.01 else 'inspect first-key/self/recent3 masses', claim_limit='Single-source H23-conditioned, single-head QK probability/readout similarity; not bidirectional, not V similarity, not full-QKV-head equivalence; exploratory addendum independently fresh-tested.') except Exception as exc: status.update(state='stopped',error_type=type(exc).__name__,error=str(exc));(root/'conditional-error.txt').write_text(traceback.format_exc());raise finally: status.update(elapsed_seconds=time.monotonic()-started,peak_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, shared_budget_seconds_including_60s_early_reserve=reserve+time.monotonic()-started) write_json(root/'conditional-status.json',status) if __name__=='__main__':main()