src/solomon/heads.py

"""Final normalized feature extraction, preserving the qualified CUDA engine."""
import hashlib,json
from pathlib import Path
import numpy as np

def sha(path):return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def token_ids_for(engine,state,block,n):
    text=engine._render(state['parts'],block);raw=engine.t.encode(text,add_special_tokens=False);ids=[]
    for c in 'ABCDEFGHIJ'[:n]:
        ext=engine.t.encode(text+c,add_special_tokens=False)
        if ext[:-1]!=raw or len(ext)!=len(raw)+1:raise ValueError('unstable answer token continuation')
        ids.append(ext[-1])
    return ids

def capture(engine,state,block,n):
    """Prehook sees precisely the post-final-norm input used by the vocabulary head."""
    captured=[]
    def hook(module,args):
        h=args[0]
        if h.ndim!=2 or h.shape[0]!=1:raise ValueError('unexpected final head input shape')
        captured.append(h.detach().float().cpu().numpy()[0].copy())
    handle=engine.model.lm_head.register_forward_pre_hook(hook)
    try:result=engine.ask(state,block,n,execution='cached')
    finally:handle.remove()
    if len(captured)!=1:raise ValueError('expected exactly one vocabulary projection')
    ids=token_ids_for(engine,state,block,n)
    if not np.isfinite(captured[0]).all():raise ValueError('nonfinite hidden features')
    return captured[0],ids,result

def copy_letter_head(engine):
    ids=[]
    for c in 'ABCDEFGHIJ':
        pieces=engine.t.encode(c,add_special_tokens=False)
        if len(pieces)!=1:raise ValueError('letters must be one token')
        ids.append(pieces[0])
    head=engine.model.lm_head;w=head.weight.detach()[ids].float().cpu().numpy().copy();b=getattr(head,'bias',None)
    return {'weight':w,'bias':np.zeros(10,np.float32) if b is None else b.detach()[ids].float().cpu().numpy().copy(),'token_ids':np.array(ids,np.int64),'vocabulary_size':np.array(head.weight.shape[0],np.int64),'bias_present':np.array(b is not None)}

def atomic_json(path,value):
    path=Path(path);tmp=path.with_suffix(path.suffix+'.tmp');tmp.write_text(json.dumps(value,sort_keys=True,indent=2));tmp.replace(path)

def write_document(path,features,token_ids,results,metadata,identity):
    """NPZ first, then checksummed completion JSON. No cumulative shard rewrite."""
    path=Path(path);n=len(features)
    if not n or not(len(token_ids)==len(results)==len(metadata)==n):raise ValueError('document arrays must align')
    h=np.asarray(features,np.float32);logits=np.zeros((n,10),np.float32);tokens=np.full((n,10),-1,np.int64);widths=[]
    if h.ndim!=2 or not np.isfinite(h).all():raise ValueError('invalid document feature matrix')
    rows=[]
    for i,(ids,r,m) in enumerate(zip(token_ids,results,metadata)):
        values=np.asarray(r['letter_logits'],np.float32);width=len(ids)
        if not 2<=width<=10 or values.shape!=(width,) or not np.isfinite(values).all():raise ValueError('invalid teacher row')
        logits[i,:width]=values;tokens[i,:width]=ids;widths.append(width)
        rows.append({**m,'n':width,'mass':float(r['mass']),'top_is_letter':bool(r['top_is_letter']),'prompt_tokens':int(r['prompt_tokens']),'branch_tokens':int(r['branch_tokens']),'reused_prefix_tokens':int(r['reused_prefix_tokens'])})
    tmp=path.with_suffix('.npz.tmp')
    with tmp.open('wb') as f:np.savez(f,hidden=h,teacher_logits=logits,token_ids=tokens,widths=np.asarray(widths,np.int64))
    tmp.replace(path)
    report={'complete':True,'identity':identity,'rows':rows,'npz_sha256':sha(path),'feature_shape':list(h.shape),'feature_kind':'final-normalized-lm-head-input','storage':'uncompressed-npz-float32'}
    atomic_json(path.with_suffix('.json'),report)
    return report

def validate_document(path,identity,expected_ids):
    path=Path(path);meta=path.with_suffix('.json')
    if not path.exists() or not meta.exists():return False
    r=json.loads(meta.read_text())
    if r.get('complete')is not True or r.get('identity')!=identity or r.get('npz_sha256')!=sha(path) or [x['id'] for x in r['rows']]!=list(expected_ids):raise ValueError('resumed document identity mismatch')
    return True

class TrainedHeads:
    """Validated saved answer/correctness heads. No backbone or label access."""
    def __init__(self,directory,*,expected_report_sha256=None,expected_weights_sha256=None):
        directory=Path(directory);report_path=directory/'report.json';weights_path=directory/'weights.npz'
        self.report_sha256=sha(report_path);self.weights_sha256=sha(weights_path)
        if expected_report_sha256 is not None and self.report_sha256!=expected_report_sha256:raise ValueError('selected head report changed')
        if expected_weights_sha256 is not None and self.weights_sha256!=expected_weights_sha256:raise ValueError('selected head weights changed')
        self.report=json.loads(report_path.read_text())
        if self.report['weights_sha256']!=self.weights_sha256:raise ValueError('trained head weight checksum mismatch')
        self.models={}
        with np.load(weights_path,allow_pickle=False) as archive:
            for kind,entries in self.report['models'].items():
                self.models[kind]={}
                for key,fields in entries.items():
                    model={name:archive[value['array']].astype(np.float64) if isinstance(value,dict) and set(value)=={'array'} else value for name,value in fields.items()}
                    if any(not np.isfinite(value).all() for value in model.values() if isinstance(value,np.ndarray)):raise ValueError('nonfinite trained head')
                    if kind=='heads':
                        if model['weight'].ndim!=2 or model['weight'].shape[0]!=10 or model['bias'].shape!=(10,):raise ValueError('invalid answer head shape')
                    elif kind=='correctness':
                        if model['mean'].ndim!=1 or model['scale'].shape!=model['mean'].shape or np.any(model['scale']<=0) or model['weights'].shape!=(len(model['mean'])+1,):raise ValueError('invalid correctness head shape')
                    else:raise ValueError('unknown trained model kind')
                    self.models[kind][key]=model
        if not self.models.get('heads') or not self.models.get('correctness'):raise ValueError('both trained answer and correctness heads required')

    def answer(self,hidden,key,n):
        if key not in self.models['heads']:raise ValueError('no trained answer head for '+str(key))
        if type(n)is not int or not 2<=n<=10:raise ValueError('invalid head output width')
        h=np.asarray(hidden,np.float64);model=self.models['heads'][key]
        if h.shape!=(model['weight'].shape[1],) or not np.isfinite(h).all():raise ValueError('invalid hidden state for answer head')
        return (model['weight']@h+model['bias'])[:n]

    def correctness(self,task,branches):
        from scipy.special import expit
        if task not in self.models['correctness'] or not branches:raise ValueError('no trained whole-answer correctness head')
        hidden=np.asarray([r['hidden'] for r in branches],np.float64);probs=[]
        if hidden.ndim!=2 or not np.isfinite(hidden).all():raise ValueError('invalid correctness hidden representations')
        for row in branches:
            p=np.asarray(row['probabilities'],float)
            if p.ndim!=1 or len(p)<2 or not np.isfinite(p).all() or np.any(p<0) or abs(p.sum()-1)>1e-6:raise ValueError('invalid correctness answer distribution')
            probs.append(p)
        feature=np.r_[hidden.mean(0),min(float(p.max()) for p in probs),min(float(np.sort(p)[-1]-np.sort(p)[-2]) for p in probs),max(float(-np.sum(p*np.log(np.maximum(p,1e-300)))/np.log(len(p))) for p in probs),float(len(branches)),float(sum(len(p) for p in probs))]
        model=self.models['correctness'][task]
        if feature.shape!=model['mean'].shape:raise ValueError('whole-answer confidence feature width mismatch')
        value=float(expit(np.r_[1.,(feature-model['mean'])/model['scale']]@model['weights']))
        return {'reliability':value,'features':feature.tolist(),'task':task,'target':'whole_return_exact_correctness','calibration_established':False,'heads_sha256':self.weights_sha256}


def semantic_head_key(task,block):
    from solomon.engine_contract import SUFFICIENCY,SINGLE_R,SINGLE_S,ORDERED_R,ORDERED_S,LABEL,BOOLEAN_TASK
    if task not in ('boolean','single','ordered','multilabel','entity'):raise ValueError('explicit answering task required')
    if block.startswith(SUFFICIENCY) and task in ('single','ordered'):return task+'/sufficiency3'
    for prefix,t,form in ((SINGLE_R,'single','R'),(SINGLE_S,'single','S'),(ORDERED_R,'ordered','R'),(ORDERED_S,'ordered','S')):
        if block.startswith(prefix) and task==t:return task+'/choice'+form
    if block.startswith(LABEL) and task=='multilabel':return 'multilabel/state4'
    if block.startswith(BOOLEAN_TASK):
        if task in ('boolean','entity'):return task+'/state4'
        if task=='ordered':return 'ordered/threshold4'
    raise ValueError('task block has no trained semantic head mapping')


from solomon.engine_reasoning import ReasoningEngine
class HeadsEngine(ReasoningEngine):
    """Question-only backbone with trained small heads; native LM head retained for reasoning.

    Wrap a complete readout in task_context(task), or supply ask(head_key=...).
    Answer projection is deliberately evaluated in CPU float64 on the exact copied
    float32 hidden vector to match saved-feature evaluation; no speedup is claimed.
    """
    def __init__(self,heads_dir,*,expected_base_fingerprint,expected_report_sha256=None,expected_weights_sha256=None,**kwargs):
        import threading
        self.heads=TrainedHeads(heads_dir,expected_report_sha256=expected_report_sha256,expected_weights_sha256=expected_weights_sha256)
        super().__init__(**kwargs)
        if self.identity['fingerprint']!=expected_base_fingerprint:raise ValueError('trained head backbone/adapter runtime mismatch')
        self.base_identity=dict(self.identity);self._head_task=None;self._head_lock=threading.RLock()
        self.identity={**self.identity,'base_fingerprint':expected_base_fingerprint,'trained_heads_sha256':self.heads.weights_sha256,'trained_heads_report_sha256':self.heads.report_sha256,'heads_runtime_sha256':sha(__file__),'answer_projection':'trained-semantic-head-float64-cpu'}
        self.identity.pop('fingerprint',None);self.identity['fingerprint']=hashlib.sha256(json.dumps(self.identity,sort_keys=True).encode()).hexdigest()

    def task_context(self,task):
        from contextlib import contextmanager
        @contextmanager
        def context():
            with self._head_lock:
                previous=self._head_task;self._head_task=task
                try:yield self
                finally:self._head_task=previous
        return context()

    def prefill(self,document):
        with self._head_lock:return super().prefill(document)

    def ask(self,state,block,n_letters,execution='cached',head_key=None):
        import copy
        with self._head_lock:
            key=head_key or semantic_head_key(self._head_task,block)
            if key not in self.heads.models['heads']:raise ValueError('missing trained head '+key)
            if self.placement!='question':raise ValueError('trained heads require question-only adaptation')
            if execution not in ('cached','full'):raise ValueError('invalid execution')
            torch=self.torch;p=state['prefix_tokens'];previous=self.ctx['start']
            try:
                with torch.inference_mode():
                    text=self._render(state['parts'],block);ids,embeds,pos,_=self._encode(state['parts'],text,state['features'])
                    if not torch.equal(ids[:,:p],state['prefix_ids']):raise ValueError('head question prefix mismatch')
                    self.ctx['start']=(0 if execution=='cached' or self.placement=='full' else p) if self.adapter else None
                    if execution=='cached':h=self.lm(inputs_embeds=embeds[:,p:],position_ids=pos[...,p:],past_key_values=copy.deepcopy(state['cache']),use_cache=True).last_hidden_state
                    else:h=self.lm(inputs_embeds=embeds,position_ids=pos,use_cache=False).last_hidden_state
                    hidden=h[0,-1].float().cpu().numpy().copy()
                logits=self.heads.answer(hidden,key,n_letters);probs=np.exp(logits-logits.max());probs/=probs.sum()
                return {'letter_logits':logits,'probabilities':probs,'prediction':int(probs.argmax()),'hidden':hidden.tolist(),'head_key':key,'mass':None,'top_is_letter':None,'mass_available':False,'execution':execution,'fallback':'','prompt_tokens':int(ids.shape[1]),'branch_tokens':int(ids.shape[1])-p,'input_tokens':int(ids.shape[1])-p,'reused_prefix_tokens':p if execution=='cached' else 0,'generation_tokens':0,'fingerprint':self.identity['fingerprint']}
            finally:self.ctx['start']=previous

    def estimate(self,state,block,n,stage='fast',max_new_tokens=512):
        if stage!='fast' or not any('image' in part for part in state['parts']):
            return super().estimate(state,block,n,stage=stage,max_new_tokens=max_new_tokens)
        if type(n)is not int or not 2<=n<=10:raise ValueError('invalid letter count')
        # Reuse prefetched vision features; no transformer branch is evaluated.
        with self._head_lock,self.torch.inference_mode():
            ids,_,_,_=self._encode(state['parts'],self._render(state['parts'],block),state['features'])
            prefix=state['prefix_tokens']
            if not self.torch.equal(ids[:,:prefix],state['prefix_ids']):raise ValueError('image budget prefix mismatch')
            return {'input_tokens':int(ids.shape[1])-prefix,'generated_tokens':0,'branches':1,
                    'full_prompt_tokens':int(ids.shape[1]),'reused_prefix_tokens':prefix}

    def predict_correctness(self,task,branches):return self.heads.correctness(task,branches)

    def reason(self,*args,**kwargs):
        with self._head_lock:return super().reason(*args,**kwargs)