File size: 6,044 Bytes
c49eca9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """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)
|