sources/qualification-snapshot/scripts/validate_caption_run.py

"""Independent validation of append-only live caption events and host receipts."""
import hashlib,struct,math
from pathlib import Path
from probe_native_stream import token_table,token_oracle
from score_stream_latency import score

def require(value,message):
 if not value:raise ValueError(message)

def validate(rows,observations,case,tokenizer,*,frozen_segments=None,expected_ids=None):
 require(len(rows)==len(observations) and len(rows)>=2,'observer event coverage')
 require(rows[0]['kind']=='caption_stream_ready' and rows[-1]['kind']=='caption_stream_complete','caption envelope')
 allowed={'caption_stream_ready','caption_segment_start','caption_window','caption_text_delta','caption_segment_end','caption_stream_complete'}
 require(all(r['kind'] in allowed for r in rows),'unknown caption event')
 require(sum(r['kind']=='caption_stream_ready' for r in rows)==sum(r['kind']=='caption_stream_complete' for r in rows)==1,'duplicate envelope')
 final=rows[-1];require(final['status']=='complete' and final['audio_samples']==case['samples'] and final['input_sha256']==case['sha256'],'input identity/completion')
 table=token_table(tokenizer);allids=[];local=[];segments=[];active=None;previous_end=0;oracle=[];actual=[];deltas=[];last_time=-1
 for i,(r,o) in enumerate(zip(rows,observations)):
  require(o['event_index']==i and o['kind']==r['kind'] and o.get('index')==r.get('index'),'host observation mapping')
  at=o['received_seconds_since_feeder_start']
  if r['kind']=='caption_stream_ready':continue
  require(type(at) in (int,float) and math.isfinite(at) and at>=max(0,last_time),'receipt clock')
  last_time=at
  if r['kind']=='caption_segment_start':
   require(active is None and r['segment']==len(segments),'segment ordering')
   active=dict(id=r['segment'],start_sample=r['start_sample'],decision_sample=r['decision_sample'])
   require(previous_end<=active['start_sample']<=active['decision_sample'] and active['start_sample']==max(previous_end,active['decision_sample']-6400) and active['decision_sample']%512==0,'activation/preroll')
   local=[]
  elif r['kind']=='caption_window':
   require(active is not None and r['segment']==active['id'],'window outside active session')
   require(r['index']==len(allids) and r['segment_index']==len(local),'global/local clock')
   token=r['token_id'];require(type(token) is int and 0<=token<table['model_count'] and token!=151645,'invalid/suppressed token')
   require(r['source_dependency_sample']==active['start_sample']+len(local)*1280+40,'source window clock')
   require(r['available_source_sample']>=active['decision_sample'] and r['available_source_sample']<=case['samples'],'source availability')
   require(type(r['eof_drain']) is bool and r['eof_drain']==(r['source_dependency_sample']>r['available_source_sample']),'EOF right padding')
   require(0<=r['cache_bytes']<=220_000_000 and r['rolling_trims']>=0,'cache bound')
   require(r['work_seconds']>=0 and r['queue_seconds'] is not None and r['queue_seconds']>=0,'live timing values')
   local.append(r);allids.append(token)
  elif r['kind']=='caption_text_delta':
   require(active is not None and r['segment']==active['id'] and type(r['text']) is str,'text outside session')
   actual.append(r['text']);deltas.append(dict(text=r['text'],seconds=at))
  elif r['kind']=='caption_segment_end':
   require(active is not None and r['segment']==active['id'] and r['start_sample']==active['start_sample'],'end identity')
   length=r['end_sample']-active['start_sample'];require(length>0,'empty segment')
   count=max(0,(length+(14*1280 if r['final_eof'] else 0)-40)//1280+1)
   require(len(local)==r['windows']==count,'segment complete windows')
   require(not r['final_eof'] or r['end_sample']==case['samples'],'nonfinal source EOF')
   require(r['final_eof'] or (r['end_sample']%512==0 and not any(w['eof_drain'] for w in local)),'synthetic nonfinal EOF')
   consumed=min(length,(count-1)*1280+40) if count else 0
   require(r['unconsumed_samples']==length-consumed,'explicit unconsumed tail')
   require(all(w['available_source_sample']<=r['end_sample'] for w in local),'read beyond segment boundary')
   text=token_oracle([w['token_id'] for w in local],table)['text']
   if text:oracle.append(text)
   segments.append(dict(active,end_sample=r['end_sample'],final_eof=r['final_eof'],windows=count,unconsumed_samples=r['unconsumed_samples']))
   previous_end=r['end_sample'];active=None
 require(active is None and len(segments)==final['segments'] and len(allids)==final['emissions'],'final accounting')
 require(''.join(actual)=='\n'.join(oracle),'independent text oracle')
 require(sum(len(t.encode('utf8')) for t in actual)==final['text_bytes'],'text byte count')
 require(hashlib.sha256(b''.join(struct.pack('<I',i) for i in allids)).hexdigest()==final['generated_ids_sha256'],'token digest')
 require(sum(s['end_sample']-s['start_sample'] for s in segments)==final['selected_samples'] and sum(s['unconsumed_samples'] for s in segments)==final['unconsumed_selected_samples'],'selected input accounting')
 require(final['max_retained_input_samples']<=6912,'bounded input')
 if frozen_segments is not None:
  require(len(segments)==len(frozen_segments),'frozen segment count')
  for got,expected in zip(segments,frozen_segments):
   require(all(got[k]==expected[k] for k in ('id','start_sample','end_sample','decision_sample','final_eof')),'frozen segment decision')
 if expected_ids is not None:require(allids==expected_ids,'exact prior candidate IDs')
 scored=score(case,deltas,actual_arrival_paced=True,timing_basis='Independent host receipt of flushed JSONL since paced20ms producer start; all source words retained')
 windows=[r for r in rows if r['kind']=='caption_window']
 return dict(status='complete',summary=final,score=scored,segments=segments,
  all_ids_exact_expected=(True if expected_ids is not None else None),independent_tokenizer_oracle=True,
  max_queue_seconds=max((w['queue_seconds'] for w in windows),default=0.),
  no_synthetic_nonfinal_eof=True,retention_bounded=True,release_qualification=False)