| |
| """Frozen continuous-stream probe with raw output, arrival timing, RSS and quality.""" |
| import argparse |
| import codecs |
| import contextlib |
| import hashlib |
| import json |
| import math |
| import os |
| from pathlib import Path |
| import platform |
| import signal |
| import statistics |
| import struct |
| import subprocess |
| import sys |
| import threading |
| import time |
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT)) |
| from run_native_probe import inspect_prefix, parse_time, score, PREFIX_ARTIFACT_BYTES, PREFIX_HEADER_BYTES, PREFIX_PAYLOAD_BYTES |
|
|
| def digest(path): |
| h=hashlib.sha256() |
| with Path(path).open('rb') as f: |
| for b in iter(lambda:f.read(1048576),b''): h.update(b) |
| return h.hexdigest() |
|
|
| def require(condition, message): |
| if not condition: |
| raise ValueError(message) |
|
|
|
|
| def finite(value, name, signed=False): |
| require(type(value) in (int, float) and math.isfinite(value) |
| and (signed or value >= 0), f"invalid numeric field {name}") |
| return value |
|
|
|
|
| def integer(value, name, minimum=0): |
| require(type(value) is int and value >= minimum, f"invalid integer field {name}") |
| return value |
|
|
|
|
| def token_table(path): |
| """Independent reader of the frozen A8TOK001 byte-piece table; no model load.""" |
| data = Path(path).read_bytes() |
| require(168 <= len(data) <= 16*1024*1024 and data[:8] == b"A8TOK001", "invalid token table") |
| count, model_count, byte_count, *special = struct.unpack_from("<8I", data, 8) |
| require(5 <= count <= model_count <= 262144, "invalid token table dimensions") |
| end = 168 + (count+1)*4 |
| require(end+count+byte_count == len(data), "token table exact length differs") |
| offsets = struct.unpack_from(f"<{count+1}I", data, 168) |
| flags, pieces = data[end:end+count], data[end+count:] |
| require(offsets[0] == 0 and offsets[-1] == byte_count, "invalid token offsets") |
| require(all(0 < b-a <= 4096 for a,b in zip(offsets, offsets[1:])), "invalid token piece spans") |
| require(set(flags) <= {0,1} and len(set(special)) == 5 |
| and all(i < count and flags[i] == 1 for i in special), "invalid special IDs") |
| return dict(count=count, model_count=model_count, offsets=offsets, flags=flags, pieces=pieces, special=special) |
|
|
|
|
| def token_oracle(ids, table): |
| """Use Python's independent incremental UTF-8 decoder and identical ID filtering.""" |
| decoder = codecs.getincrementaldecoder("utf-8")("replace") |
| skipped, visible, deltas, stopped = [0]*5, 0, [], False |
| for index, token in enumerate(ids): |
| integer(token, "token ID") |
| require(token <= 0xffffffff, "token ID exceeds u32") |
| if stopped: |
| continue |
| if token in table['special']: |
| which = table['special'].index(token) |
| skipped[which] += 1 |
| if which == 3: |
| text = decoder.decode(b'', final=True) |
| if text: |
| deltas.append((index, text, False)) |
| stopped = True |
| continue |
| visible += 1 |
| if token >= table['count'] or table['flags'][token]: |
| continue |
| a,b = table['offsets'][token:token+2] |
| text = decoder.decode(table['pieces'][a:b], final=False) |
| if text: |
| deltas.append((index, text, False)) |
| if not stopped: |
| tail = decoder.decode(b'', final=True) |
| if tail: |
| deltas.append((len(ids), tail, True)) |
| return dict(text=''.join(text for _,text,_ in deltas), visible=visible, |
| skipped=skipped, deltas=deltas) |
|
|
|
|
| def output_paths(output): |
| output = Path(os.path.abspath(output)) |
| output = output.parent.resolve() / output.name |
| require(output.suffix == '.json', 'output must have .json suffix') |
| paths = dict(report=output, raw=output.with_suffix('.jsonl'), |
| timing=output.with_suffix('.time.txt'), errors=output.with_suffix('.stderr.txt'), |
| feeder=output.with_suffix('.feeder.jsonl')) |
| require(len(set(paths.values())) == len(paths), 'output paths overlap') |
| for path in paths.values(): |
| require(not path.exists() and not path.is_symlink(), f'output already exists: {path}') |
| return paths |
|
|
|
|
| def load_input_case(manifest, case_id): |
| """Read either the frozen case collection or one raw-F32LE stress case.""" |
| data = json.loads(Path(manifest).read_text()) |
| require(isinstance(data, dict), 'input manifest must be an object') |
| if 'cases' in data: |
| require(data.get('format') == 'mono_f32le' |
| and type(data.get('sample_rate')) is int and data['sample_rate'] == 16000, |
| 'input format differs') |
| require(isinstance(data['cases'], list), 'input cases must be a list') |
| candidates = data['cases'] |
| else: |
| |
| |
| require('format' not in data or data['format'] == 'mono_f32le', 'input format differs') |
| require('sample_rate' not in data or (type(data['sample_rate']) is int |
| and data['sample_rate'] == 16000), 'input sample rate differs') |
| candidates = [data] |
| require(all(isinstance(row, dict) and isinstance(row.get('id'), str) |
| for row in candidates), 'invalid input case entries') |
| require(len({row['id'] for row in candidates}) == len(candidates), 'duplicate input case IDs') |
| matches = [row for row in candidates if row['id'] == case_id] |
| require(len(matches) == 1, 'missing or duplicate input case') |
| case = matches[0] |
| samples = integer(case.get('samples'), 'manifest audio samples', 1) |
| require(finite(case.get('seconds'), 'manifest audio seconds') == samples/16000, |
| 'input duration differs') |
| require(case.get('language') in ('en', 'zh'), 'unsupported input language') |
| require(isinstance(case.get('path'), str) and Path(case['path']).is_absolute(), |
| 'input path must be absolute') |
| checksum = case.get('sha256') |
| require(isinstance(checksum, str) and len(checksum) == 64 |
| and set(checksum) <= set('0123456789abcdef'), 'invalid input checksum') |
| return case |
|
|
|
|
| def gear_profile(gear, batch, delay_ms=480, first_batch=None): |
| require(type(gear) is int and gear in (4, 6, 8), 'unsupported expected gear') |
| require(type(delay_ms) is int and delay_ms in (240,480), 'unsupported transcription delay') |
| if delay_ms == 240: |
| require(gear == 4 and type(batch) is int and batch in (1,2,4) |
| and type(first_batch) is int and first_batch == 1, |
| '240ms candidate requires gear4, batch1/2/4 and explicit first batch1') |
| else: |
| require(first_batch is None, '480ms historical profiles use their default first batch') |
| require(type(batch) is int and batch in ((8, 16) if gear == 4 else (4, 8, 16)), |
| 'unsupported expected batch') |
| delay=delay_ms//(20*gear) |
| return dict(gear=gear, period=320*gear, clock=gear*.02, delay=delay,delay_ms=delay_ms, |
| startup=19+delay, context=1500//gear, trim=(150+gear-1)//gear, |
| first_batch=min(batch,8) if first_batch is None else first_batch, allocation=196345856+(1500//gear)*36864, |
| policy='hf_helper_80ms_native_startup' if gear == 4 else 'experimental_native_gear_derived') |
|
|
|
|
| def inspect_stream_prefix(path, language, assets, profile): |
| """Keep the old prefix reader unchanged; independently validate delay-bound V2.""" |
| if profile['delay_ms'] == 480: |
| return inspect_prefix(path,language,assets,4,'a8',profile['first_batch'],1) |
| require(profile['gear'] == 4 and profile['delay'] == 3 and profile['first_batch'] == 1, |
| 'unsupported delay-bound prefix profile') |
| path=Path(path).resolve() |
| require(path.stat().st_size == PREFIX_ARTIFACT_BYTES, 'prefix artifact size differs') |
| record=dict(path=str(path),bytes=path.stat().st_size,sha256=digest(path)) |
| with path.open('rb') as stream: |
| header=stream.read(PREFIX_HEADER_BYTES) |
| require(header[:8] == b'A8PFX002', '240ms requires delay-bound V2 prefix') |
| values=struct.unpack('<25I',header[8:108]) |
| expected=(dict(en=151668,zh=151667)[language],4,3,18,16000,400,160,840,40, |
| 68,17,32,36,2048,256,16,1,4,1,1,22,128,151644,151665,151645) |
| require(values == expected, 'delay-bound prefix language/execution profile differs') |
| hashes=[header[i:i+32].hex() for i in range(108,PREFIX_HEADER_BYTES,32)] |
| require(all(h != '0'*64 for h in hashes), 'empty prefix identity hash') |
| require(hashes[1] == assets['frontend']['sha256'] |
| and hashes[2] == assets['runtime_binary']['sha256'], 'prefix frontend/math identity differs') |
| with Path(assets['model_stream']['path']).open('rb') as model: |
| model.seek(-32,2) |
| require(model.read(32).hex() == hashes[0], 'prefix model body identity differs') |
| checksum=hashlib.sha256(header) |
| remaining=record['bytes']-PREFIX_HEADER_BYTES-32 |
| while remaining: |
| block=stream.read(min(1048576,remaining)) |
| require(bool(block), 'truncated prefix body') |
| checksum.update(block);remaining-=len(block) |
| require(stream.read(32) == checksum.digest() and not stream.read(1), 'prefix footer/trailing bytes differ') |
| metadata=dict(path=str(path),body_sha256=checksum.hexdigest(),payload_bytes=PREFIX_PAYLOAD_BYTES, |
| artifact_bytes=record['bytes'],encoder_frames=68,decoder_tokens=17, |
| model_body_sha256=hashes[0],frontend_sha256=hashes[1],math_sha256=hashes[2], |
| runtime_sha256=hashes[3],layout_sha256=hashes[4],format='A8PFX002', |
| transcription_delay_ms=240,first_batch_windows=1,startup_input_tokens=22) |
| |
| return record,metadata |
|
|
|
|
| def validate(rows, case, *, expected_mode, table, prefix_metadata, |
| prior_native=None, nonrolling_reference=None, expected_batch=8, |
| expected_gear=4, model_body_sha256=None, continuous_reference=None, |
| expected_delay_ms=480,expected_first_batch=None): |
| profile = gear_profile(expected_gear, expected_batch,expected_delay_ms,expected_first_batch) |
| first_batch = profile['first_batch'] |
| period, clock, startup, context, trim = [profile[k] for k in ('period','clock','startup','context','trim')] |
| require((prefix_metadata is not None) == (expected_gear == 4), 'prefix route differs from gear contract') |
| allowed = {'native_stream_ready', 'native_stream_complete', 'native_stream_window', |
| 'native_stream_batch', 'native_text_delta'} |
| require(rows and all(isinstance(r, dict) and r.get('kind') in allowed for r in rows), 'unknown/malformed event') |
| ready = [r for r in rows if r['kind'] == 'native_stream_ready'] |
| final = [r for r in rows if r['kind'] == 'native_stream_complete'] |
| require(len(ready) == len(final) == 1 and rows[0] is ready[0] and rows[-1] is final[0], 'ready/final event coverage differs') |
| ready, final = ready[0], final[0] |
| expected = dict(gear=expected_gear, workers=4, batch_windows=expected_batch, row_tasks=1, delay_tokens=profile['delay'], |
| prefix_used=prefix_metadata is not None, format='mono_16000_f32le', retains_transcript=False, |
| rolling_context_tokens=context, rolling_interval_tokens=trim, |
| rolling_stable_prefix_tokens=16, microphone_test=False, |
| production_accuracy_approved=False, input_mode=expected_mode) |
| for key,value in expected.items(): |
| require(ready.get(key) == value and type(ready[key]) is type(value), f'requested profile differs: {key}') |
| for target in (ready,final): |
| for key,value in dict(transcription_delay_ms=expected_delay_ms,first_batch_windows=first_batch,cache_digest_enabled=False).items(): |
| if expected_delay_ms != 480 or key in target: |
| require(target.get(key) == value and type(target[key]) is type(value), f'delay/timing profile differs: {key}') |
| new_fields = dict(token_period_ms=20*expected_gear, samples_per_token=period, |
| startup_input_tokens=startup, eof_drain_tokens=11, rolling_policy=profile['policy'], |
| experimental_gear_policy=expected_gear != 4, upstream_realtime_gear_supported=expected_gear == 4) |
| for key,value in new_fields.items(): |
| if expected_gear != 4 or expected_delay_ms != 480 or key in ready: |
| require(ready.get(key) == value and type(ready[key]) is type(value), f'gear metadata differs: {key}') |
| for key,value in dict(gear=expected_gear,token_period_ms=20*expected_gear,rolling_policy=profile['policy'],experimental_gear_policy=expected_gear != 4).items(): |
| if expected_gear != 4 or key in final: |
| require(final.get(key) == value and type(final[key]) is type(value), f'final gear metadata differs: {key}') |
| if expected_gear != 4 or expected_batch != 8 or 'first_batch_windows' in ready: |
| require(type(ready.get('first_batch_windows')) is int and ready['first_batch_windows'] == first_batch, 'initial prefix batch shape differs') |
| require(ready['input'] == ('-' if expected_mode == 'external_stream' else case['path']), 'input route differs') |
| expected_prefix = {k:prefix_metadata[k] for k in ('path','body_sha256','artifact_bytes')} if prefix_metadata else None |
| require(ready['prefix_cache'] == expected_prefix, 'prefix metadata differs') |
| expected_model = prefix_metadata['model_body_sha256'] if prefix_metadata else model_body_sha256 |
| require(isinstance(expected_model,str) and len(expected_model) == 64 |
| and ready['model_body_sha256'] == expected_model, 'model identity differs') |
| require(final['status'] == 'complete' and final['all_source_audio_observed'] is True |
| and final['input_mode'] == expected_mode and final['production_accuracy_approved'] is False, |
| 'invalid complete/source coverage record') |
| require(integer(final['audio_samples'], 'audio samples', 1) == case['samples'], 'audio count differs') |
| require(final['audio_seconds'] == case['samples']/16000 == case['seconds'], 'audio duration differs') |
| windows = [r for r in rows if r['kind'] == 'native_stream_window'] |
| batches = [r for r in rows if r['kind'] == 'native_stream_batch'] |
| deltas = [r for r in rows if r['kind'] == 'native_text_delta'] |
| count = (case['samples']+period-1)//period+11 |
| require(len(windows) == count == final['emitted_tokens'], 'emission coverage differs') |
| ids = [r['token_id'] for r in windows] |
| require(all(type(i) is int and 0 <= i < table['model_count'] and i != table['special'][3] for i in ids), 'invalid/suppressed output token ID') |
| oracle = token_oracle(ids, table) |
| require(hashlib.sha256(b''.join(struct.pack('<I',i) for i in ids)).hexdigest() == final['generated_ids_sha256'], 'ID digest differs') |
| previous_time, cached, dropped, trims = -1.0, 0, 0, 0 |
| for i,r in enumerate(windows): |
| incoming = startup if i == 0 else 1 |
| if cached+incoming > context: |
| drop = cached+incoming-context+trim-1 |
| cached -= drop |
| dropped += drop |
| trims += 1 |
| cached += incoming |
| require(r['index'] == i and r['absolute_text_position'] == startup+i |
| and r['absolute_encoder_position'] == (startup+i)*expected_gear, 'absolute clock differs') |
| require(r['text_position'] == cached and r['encoder_position'] == cached*expected_gear |
| and r['rolling_dropped_tokens'] == dropped and r['rolling_trim_count'] == trims, |
| 'rolling clock/trim policy differs') |
| finite(r['seconds'], 'window timestamp') |
| require(r['seconds'] >= previous_time, 'window time reversed') |
| previous_time = r['seconds'] |
| require(count+startup-1 <= context or trims > 0, 'long case did not cross a rolling trim') |
| elapsed = finite(final['elapsed_seconds'], 'elapsed seconds') |
| index, prior_finished, allocations = 0, 0.0, [] |
| for r in batches: |
| require(r['start_window'] == index, 'batch coverage gap') |
| end = integer(r['end_window_exclusive'], 'batch end', 1) |
| require(end == min(index+(first_batch if index == 0 else expected_batch),count), 'batch shape differs') |
| start = finite(r['started_seconds'], 'batch start') |
| finish = finite(r['finished_seconds'], 'batch finish') |
| work = finite(r['work_seconds'], 'batch work') |
| require(prior_finished <= start <= finish <= elapsed, 'batch times out of order') |
| require(math.isclose(finish-start,work,abs_tol=.02), 'batch work interval differs') |
| require(all(start <= w['seconds'] <= finish for w in windows[index:end]), 'window outside batch interval') |
| require(r['rolling_trim_count'] == windows[end-1]['rolling_trim_count'] |
| and r['decoder_cached_tokens'] == windows[end-1]['text_position'], 'batch/cache trim metadata differs') |
| allocated = integer(r['cache_allocated_bytes'], 'cache capacity', 1) |
| require(allocated <= 220_000_000, 'cache capacity exceeds bound') |
| if expected_gear != 4 or expected_delay_ms != 480: |
| require(allocated == profile['allocation'], 'derived gear cache capacity differs') |
| require(all(0 <= w['cache_logical_bytes'] <= allocated for w in windows[index:end]), 'logical cache exceeds capacity') |
| allocations.append(allocated) |
| require(r.get('initial_cache_sha256') is None, 'diagnostic cache hashing must be disabled for timing') |
| require(integer(r['retained_audio_capacity_samples'], 'audio capacity') <= 32768, 'input storage grew') |
| require(type(r['input_eof']) is bool |
| and r['input_eof'] == ((profile['delay']+end)*period+40 > case['samples']), 'invalid EOF flag') |
| deadline = min(case['samples'],(profile['delay']+end)*period+40)/16000 |
| require(r['source_audio_ready_seconds'] == deadline, 'source deadline differs') |
| relative = finite(r['relative_to_audio_clock_seconds'], 'relative audio clock', signed=True) |
| require(math.isclose(relative,start-deadline,abs_tol=1e-8), 'relative clock differs') |
| if expected_mode == 'file_paced_simulation': |
| require(start+1e-7 >= deadline, 'model ran before paced audio was available') |
| index, prior_finished = end, finish |
| require(index == count and len(set(allocations)) == 1, 'missing batches or nonconstant cache allocation') |
| require(final['peak_cache_allocated_bytes'] == allocations[0], 'peak cache allocation differs') |
| require(final['retained_audio_capacity_samples'] <= 32768, 'final input storage exceeded bound') |
| actual_deltas = [(r['index'], r['text'], r.get('eof_utf8_flush',False)) for r in deltas] |
| require(actual_deltas == oracle['deltas'], 'append-only text deltas differ from independent tokenizer oracle') |
| for r in deltas: |
| finite(r['seconds'], 'text timestamp') |
| expected_time = windows[r['index']]['seconds'] if r['index'] < count else None |
| require(r['seconds'] == expected_time if expected_time is not None |
| else windows[-1]['seconds'] <= r['seconds'] <= elapsed, 'text timestamp differs') |
| text = oracle['text'] |
| require(len(text.encode()) == final['text_bytes'] and final['visible_tokens'] == oracle['visible'] |
| and final['skipped_counts'] == oracle['skipped'], 'text/count metadata differs') |
| require(final['first_text_seconds'] == (deltas[0]['seconds'] if deltas else None) |
| and final['last_text_seconds'] == (deltas[-1]['seconds'] if deltas else None), 'text event endpoints differ') |
| reading = finite(final['read_and_window_materialization_seconds'], 'read time') |
| sleeping = finite(final['arrival_sleep_seconds'], 'sleep time') |
| waiting = finite(final['read_and_arrival_wait_seconds'], 'read/wait time') |
| total_work = finite(final['batch_work_seconds'], 'total batch work') |
| require(reading+sleeping <= waiting+1e-5 and waiting+total_work <= elapsed+1e-5, 'time accounting overlaps') |
| require(math.isclose(total_work,sum(r['work_seconds'] for r in batches),abs_tol=1e-8), 'total work differs') |
| eof = finite(final['input_eof_detected_seconds'], 'observed EOF') |
| require(eof <= elapsed and math.isclose(final['completion_after_eof_detected_seconds'],elapsed-eof,abs_tol=1e-8) |
| and math.isclose(final['completion_relative_to_audio_clock_seconds'],elapsed-case['seconds'],abs_tol=1e-8), 'EOF/nominal clock accounting differs') |
| active = [r for r in batches if r['start_window'] > 0 and not r['input_eof'] |
| and r['end_window_exclusive']-r['start_window'] == expected_batch] |
| work = [r['work_seconds']/expected_batch for r in active] |
| queue = [max(0,r['relative_to_audio_clock_seconds']) for r in active] |
| report = dict(ready=ready,summary=final,text=text,generated_token_ids=ids, |
| independent_tokenizer_oracle_passed=True,constant_cache_allocation_verified=True, |
| rolling_trims=trims,first_trim_index=next((r['index'] for r in windows if r['rolling_trim_count']),None), |
| source_active_batch_windows=expected_batch,first_batch_windows=first_batch, |
| source_active_clock_seconds=clock,frame_gear=expected_gear, |
| additional_batch_buffering_seconds=(expected_batch-1)*clock, |
| source_active_p95_work_seconds_per_window=sorted(work)[math.ceil(.95*len(work))-1] if work else None, |
| source_active_p95_normalized_work_rtf=sorted(work)[math.ceil(.95*len(work))-1]/clock if work else None, |
| source_active_batches=len(active),source_active_over_budget=sum(x>clock for x in work), |
| max_work_seconds_per_window=max(work,default=0),median_work_seconds_per_window=statistics.median(work) if work else None, |
| first_eight_mean_queue_seconds=statistics.mean(queue[:8]) if queue else None, |
| last_eight_mean_queue_seconds=statistics.mean(queue[-8:]) if queue else None, |
| max_queue_seconds=max(queue,default=0), |
| last_text_relative_to_nominal_audio_clock_seconds=(final['last_text_seconds']-case['seconds']) if deltas else None) |
| if expected_delay_ms != 480: |
| report.update(transcription_delay_ms=expected_delay_ms, |
| new_hard_latency_diagnostic=dict(maximum_seconds=1.,first_text_from_audio_clock_seconds=final['first_text_seconds'], |
| first_text_passed=final['first_text_seconds'] is not None and final['first_text_seconds'] <= 1., |
| speech_onset_latency_measured=False,aligned_content_latency_measured=False,qualified=False)) |
| if case.get('reference') is not None: |
| require(prior_native is not None, 'missing pinned prior-clip reference') |
| report['score'] = score(case['reference'],text,case['language']) |
| by_id = {r['id']:r for r in prior_native['fixtures']} |
| require(len(by_id) == len(prior_native['fixtures']), 'duplicate prior fixture IDs') |
| joiner = ' ' if case['language'] == 'en' else '' |
| reference_native = joiner.join(by_id[r['id']]['raw_result']['final_text'] for r in case['segments']) |
| report['prior_independent_clip_text'] = reference_native |
| report['prior_independent_clip_score'] = score(case['reference'],reference_native,case['language']) |
| report['quality_scope'] = 'Concatenated development clips with 0.3 s gaps versus independent clip outputs; not a controlled original rolling-model comparison or held-out quality approval.' |
| if expected_gear == 4 and expected_delay_ms == 480 and case['id'] in ('en0-repeat-30s','en0-short'): |
| require(nonrolling_reference is not None, 'missing pinned nonrolling reference') |
| cut = report['first_trim_index'] |
| if cut is None: |
| cut = len(ids) |
| report['all_pretrim_ids_match_reference'] = ids[:cut] == nonrolling_reference['generated_token_ids'][:cut] |
| report['all_ids_match_nonrolling_reference'] = ids == nonrolling_reference['generated_token_ids'] |
| report['text_matches_nonrolling_reference'] = text == nonrolling_reference['text'] |
| require(report['all_pretrim_ids_match_reference'], 'pretrim IDs differ from frozen reference') |
| if (expected_gear != 4 or expected_delay_ms != 480) and case.get('reference') is not None: |
| require(continuous_reference is not None and continuous_reference['status'] == 'complete', |
| 'missing completed gear4 continuous reference') |
| require(continuous_reference['input_case']['sha256'] == case['sha256'] |
| and continuous_reference['input_case']['samples'] == case['samples'] |
| and continuous_reference['input_case']['reference'] == case['reference'], |
| 'continuous reference input differs') |
| reference_score = continuous_reference['score'] |
| candidate_score = report['score'] |
| errors = lambda v: v['substitutions']+v['deletions']+v['insertions'] |
| report['gear4_continuous_comparison'] = dict( |
| reference_text=continuous_reference['text'],reference_score=reference_score, |
| candidate_errors=errors(candidate_score),gear4_errors=errors(reference_score), |
| additional_errors=errors(candidate_score)-errors(reference_score), |
| normalized_text_equal=candidate_score['normalized_hypothesis']==reference_score['normalized_hypothesis'], |
| token_id_parity_required=False,scope='Exact PCM/references; changed delay or trained gear/profile. Development tradeoff, not upstream realtime equivalence or held-out approval.') |
| if expected_delay_ms != 480 and nonrolling_reference is not None: |
| report['default_delay_reference_comparison']=dict(reference_text=nonrolling_reference['text'], |
| raw_text_equal=text == nonrolling_reference['text'], |
| all_ids_equal=ids == nonrolling_reference['generated_token_ids'],token_id_parity_required=False, |
| scope='Existing 480 ms same-audio diagnostic; changed transcription delay prevents requiring exact token IDs.') |
| return report |
|
|
|
|
| def geared_gate_results(report, rows, case, contract): |
| """Apply the predeclared development criteria, never production approval.""" |
| gear = report['frame_gear'] |
| require(gear in (6,8), 'geared gate requires an experimental gear') |
| clock, batch = report['source_active_clock_seconds'], report['source_active_batch_windows'] |
| result = dict(contract_status=contract['status'],production_accuracy_approved=False, |
| default_gear4_80ms_gate_unchanged=True, |
| actual_input_clock_seconds=clock, |
| batch_work_rtf=report['summary']['batch_work_seconds']/case['seconds'], |
| source_active_p95_normalized_work_rtf=report['source_active_p95_normalized_work_rtf']) |
| if 'gear4_continuous_comparison' in report: |
| key = 'english_errors_at_most' if case['language']=='en' else 'chinese_errors_at_most' |
| threshold = contract['quality_gate_for_preference_over_gear4'][key] |
| errors = report['gear4_continuous_comparison']['candidate_errors'] |
| result['quality'] = dict(candidate_errors=errors,maximum_errors=threshold, |
| passed=errors<=threshold,scope='One frozen language concatenation; both languages are required for preference, not held-out approval.') |
| if case['id']=='en0-repeat-300s' and case['samples']==4_800_000: |
| active = [r for r in rows if r['kind']=='native_stream_batch' and r['start_window']>0 |
| and not r['input_eof'] and r['end_window_exclusive']-r['start_window']==batch] |
| buckets=[] |
| for minute in range(5): |
| selected=[r for r in active if minute*60<=r['source_audio_ready_seconds']<(minute+1)*60] |
| require(selected, 'missing sustained source-minute timing population') |
| work=[r['work_seconds']/batch for r in selected] |
| queue=[max(0,r['relative_to_audio_clock_seconds']) for r in selected] |
| buckets.append(dict(source_minute=minute+1,batches=len(selected), |
| mean_work_seconds_per_window=statistics.mean(work),mean_normalized_work_rtf=statistics.mean(work)/clock, |
| p95_work_seconds_per_window=sorted(work)[math.ceil(.95*len(work))-1], |
| max_queue_seconds=max(queue),last_queue_seconds=queue[-1])) |
| gate=contract['sustained_speed_gate'] |
| growth=report['last_eight_mean_queue_seconds']-report['first_eight_mean_queue_seconds'] |
| checks=dict(whole_batch_work_rtf=result['batch_work_rtf']<=gate['batch_work_rtf_at_most'], |
| last_source_minute_mean_work_rtf=buckets[-1]['mean_normalized_work_rtf']<=gate['last_source_minute_mean_work_rtf_at_most'], |
| source_active_p95=report['source_active_p95_normalized_work_rtf']<=1, |
| queue_growth=growth<=gate['last_eight_mean_queue_growth_over_first_eight_at_most_seconds']) |
| result['sustained_speed']=dict(checks=checks,all_passed=all(checks.values()), |
| last_eight_minus_first_eight_mean_queue_seconds=growth,minute_buckets=buckets, |
| scope='Exact five-minute synthetic input and selected clock only; repeated-speech collapse remains a separate blocker.') |
| else: |
| result['sustained_speed']=dict(evaluated=False,reason='This is not the exact five-minute sustained case.') |
| return result |
|
|
|
|
| def main(): |
| p=argparse.ArgumentParser(description=__doc__) |
| p.add_argument('--binary-record',type=Path,required=True) |
| p.add_argument('--prefix',type=Path) |
| p.add_argument('--gear',type=int,choices=(4,6,8),default=4) |
| p.add_argument('--delay-ms',type=int,choices=(240,480),default=480) |
| p.add_argument('--first-batch',type=int,choices=(1,),help='Explicit first window for the 240 ms candidate only.') |
| p.add_argument('--case',required=True) |
| p.add_argument('--input-manifest',type=Path, |
| default=ROOT/'artifacts/native-continuous-inputs-v1/manifest.json') |
| p.add_argument('--mode',choices=['file-paced','stdin-paced'],default='file-paced') |
| p.add_argument('--output',type=Path,required=True) |
| p.add_argument('--batch',type=int,choices=(1,2,4,8,16),default=8,help='Historical profiles unchanged; 240 ms gear4 candidate requires 1/2/4 plus --first-batch 1.') |
| args=p.parse_args() |
| profile=gear_profile(args.gear,args.batch,args.delay_ms,args.first_batch) |
| require((args.prefix is not None) == (args.gear == 4), 'gear4 requires prefix; other gears forbid prefix') |
| paths=output_paths(args.output) |
| args.output=paths['report'] |
| require(platform.system() == 'Darwin', 'this fixed timing probe uses macOS time -l') |
| manifest=args.input_manifest.resolve() |
| case=load_input_case(manifest,args.case) |
| binary_record=json.loads(args.binary_record.read_text()) |
| binary=Path(binary_record['binary']); prefix=args.prefix.resolve() if args.prefix else None; source=Path(case['path']) |
| model=ROOT/'artifacts/native-full8-model-v1/model.a8m' |
| frontend=ROOT/'artifacts/frontend-primitive.a8m'; tokenizer=ROOT/'artifacts/native-tokenizer-v1.a8tok' |
| asset_paths=dict(binary=binary,input=source,model=model,frontend=frontend,tokenizer=tokenizer, |
| manifest=manifest,script=Path(__file__),probe_helpers=ROOT/'run_native_probe.py',binary_record=args.binary_record.resolve()) |
| if prefix: asset_paths['prefix']=prefix |
| if args.gear != 4: |
| asset_paths['gear_gate_contract']=ROOT/'reports/native-geared-stream-gate-contract.json' |
| if args.delay_ms != 480: |
| asset_paths['latency_gate_contract']=ROOT/'reports/native-streaming-latency-contract-v1.json' |
| if args.gear != 4 or args.delay_ms != 480: |
| if case.get('reference') is not None: |
| asset_paths['continuous_reference_report']=ROOT/'reports'/dict(en='native-continuous-en-105s.json',zh='native-continuous-zh-74s.json')[case['language']] |
| prior_native=nonrolling=continuous_reference=None |
| if case.get('reference') is not None: |
| asset_paths['prior_native_report']=ROOT/'reports/native_expanded_bit_b8_v1.json' |
| elif case['id'] == 'en0-repeat-30s': |
| asset_paths['nonrolling_report']=ROOT/'reports/native-paced-sdot-30s.json' |
| elif case['id'] == 'en0-short': |
| asset_paths['nonrolling_report']=ROOT/'reports/native_smoke_pool_driver_qos_v1.json' |
| assets={name:dict(path=str(path.resolve()),sha256=digest(path)) for name,path in asset_paths.items()} |
| require(assets['binary']['sha256'] == binary_record['sha256'] and assets['input']['sha256'] == case['sha256'], 'binary/input hash differs') |
| require(source.stat().st_size == case['samples']*4 and case['seconds'] == case['samples']/16000, 'input size/duration differs') |
| require(assets['model']['sha256'] == 'bf9faf53d54b6c4d27de430e2ef90c050954c81ebcf24dd9b4bd470915f7d399', 'model hash differs') |
| prefix_metadata=None |
| if prefix: |
| _, prefix_metadata=inspect_stream_prefix(prefix,case['language'],dict(frontend=assets['frontend'],runtime_binary=assets['binary'],model_stream=assets['model']),profile) |
| with model.open('rb') as stream: |
| stream.seek(-32,2); model_body=stream.read(32).hex() |
| if 'continuous_reference_report' in asset_paths: |
| continuous_reference=json.loads(asset_paths['continuous_reference_report'].read_text()) |
| gate_contract=json.loads(asset_paths['gear_gate_contract'].read_text()) if args.gear != 4 else None |
| latency_contract=json.loads(asset_paths['latency_gate_contract'].read_text()) if args.delay_ms != 480 else None |
| if latency_contract: |
| require(latency_contract['maximum_readable_text_delay_seconds'] == 1.0 |
| and latency_contract['production_accuracy_approved'] is False, 'hard latency contract differs') |
| if gate_contract: |
| policies=[v for v in gate_contract['policies'] if v['gear']==args.gear] |
| require(len(policies)==1 and policies[0] == dict(gear=args.gear,clock_seconds=profile['clock'],delay_tokens=profile['delay'],prefill_tokens=profile['startup'],context_tokens=profile['context'],trim_interval_tokens=profile['trim'],stable_prefix_tokens=16), 'prospective geared policy differs') |
| table=token_table(tokenizer) |
| if 'prior_native_report' in asset_paths: |
| prior_native=json.loads(asset_paths['prior_native_report'].read_text()) |
| if 'nonrolling_report' in asset_paths: |
| reference=json.loads(asset_paths['nonrolling_report'].read_text()) |
| if case['id'] == 'en0-repeat-30s': |
| nonrolling=reference['transcription'] |
| else: |
| selected=[r for r in reference['fixtures'] if r['id'] == '1272-128104-0000'] |
| require(len(selected) == 1, 'short reference coverage differs') |
| raw_reference=selected[0]['raw_result'] |
| nonrolling=dict(generated_token_ids=raw_reference['generated_token_ids'],text=raw_reference['final_text']) |
| expected_mode='file_paced_simulation' if args.mode == 'file-paced' else 'external_stream' |
| command=[str(binary),str(model),str(frontend),str(tokenizer),str(source) if args.mode=='file-paced' else '-',case['language'], |
| '--trace-windows'] |
| if prefix: command+=['--prefix-in',str(prefix)] |
| if args.gear != 4: command+=['--gear',str(args.gear)] |
| if args.delay_ms != 480: command+=['--delay-ms',str(args.delay_ms)] |
| if args.first_batch is not None: command+=['--first-batch',str(args.first_batch)] |
| if args.mode=='file-paced': command+=['--pace-audio'] |
| if args.batch != 8: command+=['--batch',str(args.batch)] |
| report=dict(status='running',started_at_unix=time.time(),command=command,assets=assets,input_case=case, |
| prefix_metadata=prefix_metadata,requested_profile=dict(batch_windows=args.batch,first_batch_windows=profile['first_batch'],gear=args.gear,clock_seconds=profile['clock'],transcription_delay_ms=args.delay_ms),prospective_gear_contract=gate_contract,prospective_latency_contract=latency_contract, |
| scope='Bounded native continuous session, no resets between clips. Input release follows a wall clock; not a microphone or aligned word latency test. Startup precedes ready/audio clock. Read/materialization and arrival sleeping are separate from model work. Raw failures and timing retained.',production_accuracy_approved=False) |
| with args.output.open('x') as f: json.dump(report,f,ensure_ascii=False,indent=2) |
| ready=threading.Event(); cancel=threading.Event(); thread_errors=[]; feed_records=[]; ready_events=[] |
| child=consumer=feeder=None; created_logs=[] |
| try: |
| with contextlib.ExitStack() as stack: |
| def log_file(name): |
| handle=stack.enter_context(paths[name].open('x')) |
| created_logs.append(paths[name]) |
| return handle |
| out=log_file('raw'); log_file('timing'); err=log_file('errors') |
| ledger=log_file('feeder') if args.mode=='stdin-paced' else None |
| child=subprocess.Popen(['/usr/bin/time','-l','-o',str(paths['timing']),*command],cwd=ROOT, |
| stdout=subprocess.PIPE,stderr=err,stdin=subprocess.PIPE if args.mode=='stdin-paced' else subprocess.DEVNULL, |
| start_new_session=True,bufsize=0) |
| def consume(): |
| try: |
| for line in child.stdout: |
| out.write(line.decode()); out.flush() |
| row=json.loads(line) |
| if row.get('kind')=='native_stream_ready': |
| ready_events.append(time.monotonic()) |
| ready.set() |
| except Exception as e: |
| thread_errors.append(f'consumer: {e!r}') |
| cancel.set() |
| finally: |
| ready.set() |
| child.stdout.close() |
| consumer=threading.Thread(target=consume,daemon=True); consumer.start() |
| def feed(): |
| try: |
| if not ready.wait(60): raise TimeoutError('no ready event') |
| require(len(ready_events) == 1 and not cancel.is_set(), 'no unique ready event before feeding') |
| require(child.poll() is None, 'child ended before feeding') |
| clock=time.monotonic(); samples=0 |
| with source.open('rb') as inp: |
| while chunk:=inp.read(320*4): |
| if cancel.is_set(): raise RuntimeError('feeder cancelled') |
| samples+=len(chunk)//4; deadline=samples/16000 |
| wait=deadline-(time.monotonic()-clock) |
| if wait>0 and cancel.wait(wait): raise RuntimeError('feeder cancelled') |
| before=time.monotonic()-clock |
| view=memoryview(chunk) |
| while view: |
| n=os.write(child.stdin.fileno(),view) |
| if not n: raise OSError('zero-byte pipe write') |
| view=view[n:] |
| after=time.monotonic()-clock |
| r=dict(samples=samples,deadline_seconds=deadline,before_seconds=before,after_seconds=after,write_seconds=after-before) |
| ledger.write(json.dumps(r)+'\n'); ledger.flush(); feed_records.append(r) |
| except Exception as e: |
| thread_errors.append(f'feeder: {e!r}') |
| cancel.set() |
| finally: |
| try: child.stdin.close() |
| except OSError as e: thread_errors.append(f'close stdin: {e!r}') |
| if args.mode=='stdin-paced': |
| feeder=threading.Thread(target=feed,daemon=True); feeder.start() |
| deadline=time.monotonic()+120+case['seconds']*4 |
| while child.poll() is None: |
| if thread_errors: raise ValueError(f'worker failure: {thread_errors}') |
| if time.monotonic()>deadline: raise TimeoutError('native child exceeded bounded timeout') |
| time.sleep(.05) |
| consumer.join(timeout=10) |
| if feeder: feeder.join(timeout=10) |
| require(not consumer.is_alive() and (feeder is None or not feeder.is_alive()), 'I/O thread failed to stop') |
| require(child.returncode == 0 and not thread_errors, f'child={child.returncode}; thread_errors={thread_errors}') |
| rows=[json.loads(line) for line in paths['raw'].read_text().splitlines()] |
| report.update(validate(rows,case,expected_mode=expected_mode,table=table,prefix_metadata=prefix_metadata, |
| prior_native=prior_native,nonrolling_reference=nonrolling,expected_batch=args.batch,expected_gear=args.gear,model_body_sha256=model_body,continuous_reference=continuous_reference,expected_delay_ms=args.delay_ms,expected_first_batch=args.first_batch)) |
| if gate_contract: |
| report['prospective_gear_gate_results']=geared_gate_results(report,rows,case,gate_contract) |
| report['memory']=parse_time(paths['timing'].read_text(),platform.system()) |
| if args.mode=='stdin-paced': |
| require(len(feed_records) == (case['samples']+319)//320 and feed_records[-1]['samples'] == case['samples'], 'feeder coverage differs') |
| require(all(r['before_seconds']+1e-7 >= r['deadline_seconds'] and r['after_seconds'] >= r['before_seconds'] for r in feed_records), 'feeder delivered future samples') |
| report['feeder']=dict(chunks=len(feed_records),samples=feed_records[-1]['samples'], |
| duration_seconds=feed_records[-1]['after_seconds'],max_write_seconds=max(r['write_seconds'] for r in feed_records), |
| max_delivery_lateness_seconds=max(r['after_seconds']-r['deadline_seconds'] for r in feed_records), |
| clock_scope='Feeder clock starts after receipt of native_stream_ready; no common microphone capture timestamp is assumed.') |
| report['assets_unchanged']=all(digest(r['path'])==r['sha256'] for r in assets.values()) |
| require(report['assets_unchanged'], 'pinned asset changed during run') |
| report['status']='complete' |
| except Exception as e: |
| report.update(status='failed',error=f'{type(e).__name__}: {e}') |
| cancel.set(); ready.set() |
| if child and child.poll() is None: |
| os.killpg(child.pid,signal.SIGTERM) |
| try: child.wait(timeout=5) |
| except subprocess.TimeoutExpired: os.killpg(child.pid,signal.SIGKILL); child.wait() |
| for thread in (consumer,feeder): |
| if thread: thread.join(timeout=10) |
| raise |
| finally: |
| report['finished_at_unix']=time.time() |
| report['logs']={str(path):digest(path) for path in created_logs if path.exists()} |
| args.output.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n') |
| print(json.dumps({k:report.get(k) for k in ['status','summary','rolling_trims','score','prior_independent_clip_score','source_active_over_budget','max_queue_seconds','first_eight_mean_queue_seconds','last_eight_mean_queue_seconds','memory','feeder']},ensure_ascii=False),flush=True) |
| if __name__=='__main__': main() |
|
|
