File size: 43,137 Bytes

c49eca9

#!/usr/bin/env python3
"""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:
        # The sustained-v1 single-case artifact predates collection-level format
        # metadata. The CLI/harness contract still explicitly fixes mono16k F32LE.
        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)
    # Native import additionally checks runtime identity and every cache coordinate.
    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()