sources/qualification-snapshot/scripts/probe_native_candidate.py

#!/usr/bin/env python3
"""Candidate-only native streaming harness; strict protocol, no acceptance gate.

Feeds mono16k F32LE through stdin in20ms wall-clock chunks after native readiness.
Profiles/model identity are explicit. Historical reports/probes remain unchanged.
"""
import argparse
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))
sys.path.insert(0, str(ROOT / 'scripts'))
# Normal imports: no copied tokenizer/scoring math and no namespace rebinding.
from probe_native_stream import (token_table, token_oracle, digest, require, finite,
                                 integer, load_input_case)
from native_candidate_prefix import inspect_prefix as inspect_candidate_prefix
from run_native_probe import parse_time, score


def candidate_profile(startup, delay_ms, batch, first_batch=1):
    require(startup in ('hf','realtime'), 'invalid startup policy')
    require(type(delay_ms) is int and delay_ms in (80,160,240,480), 'invalid delay')
    require(type(batch) is int and batch in (1,2,4), 'invalid candidate batch')
    require(type(first_batch) is int and first_batch == 1, 'candidate startup requires first batch1')
    delay = delay_ms//80
    return dict(gear=4,period=1280,clock=.08,delay=delay,delay_ms=delay_ms,
        startup=19+delay if startup=='hf' else 18,context=375,trim=38,
        first_batch=1,allocation=210169856,mode=startup,
        startup_policy='hf_delayed' if startup=='hf' else 'realtime18',
        eof_drain_tokens=11 if startup=='hf' else None,
        policy='hf_helper_80ms_native_startup' if startup=='hf' else 'official_realtime18_schedule_native_arithmetic')


def emission_count(samples, profile):
    integer(samples,'input samples',1)
    if profile['mode']=='hf':return (samples+1279)//1280+11
    return (samples+(profile['delay']+11)*1280-40)//1280+1


def source_end_samples(index, profile):
    integer(index,'window index')
    return (profile['startup']-18+index)*1280+40


def release_identity(path, expected_sha):
    if path is None and expected_sha is None:return None
    require(path is not None and expected_sha is not None,'release record requires path and explicit SHA256')
    path=Path(path).resolve(strict=True)
    require(digest(path)==expected_sha,'external release candidate SHA mismatch')
    record=json.loads(path.read_text())
    require(isinstance(record,dict) and bool(record),'external release candidate must be a nonempty JSON object')
    canonical=json.dumps(record,sort_keys=True,separators=(',',':'),ensure_ascii=False,allow_nan=False).encode()
    return dict(path=str(path),sha256=expected_sha,canonical_identity_sha256=hashlib.sha256(canonical).hexdigest())


def require_input_role(case, allow_heldout=False, release_candidate=None):
    if case.get('role')=='heldout':
        require(allow_heldout and release_candidate is not None,
                'heldout requires --allow-heldout and a SHA-pinned external release-candidate record')
        require(release_identity(release_candidate['path'],release_candidate['sha256'])==release_candidate,
                'external release candidate identity changed')


def validate(rows, case, *, table, profile, batch, model_body_sha256, prefix_metadata=None):
    expected_mode='external_stream'
    expected_gear=4
    expected_delay_ms=profile['delay_ms']
    expected_batch=batch
    first_batch=1
    period,clock,startup,context,trim = [profile[k] for k in ('period','clock','startup','context','trim')]
    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():
            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=profile['eof_drain_tokens'], rolling_policy=profile['policy'],
                      startup_policy=profile['startup_policy'], right_audio_pad_tokens=profile['delay']+11,
                      experimental_gear_policy=expected_gear != 4, upstream_realtime_gear_supported=expected_gear == 4)
    for key,value in new_fields.items():
        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():
        require(final.get(key) == value and type(final[key]) is type(value), f'final gear metadata differs: {key}')
    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']
    integer(final['emitted_tokens'],'emitted count',1)
    count = emission_count(case["samples"], profile)
    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):
        for field in ('index','token_id','text_position','encoder_position','absolute_text_position',
                      'absolute_encoder_position','processed_absolute_encoder_position',
                      'rolling_dropped_tokens','rolling_trim_count','cache_logical_bytes'):
            integer(r[field],field)
        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')
        require(integer(r['logit_bits'],'logit bits') <= 0xffffffff,'invalid logit encoding')
        require(isinstance(r['phase_seconds'],list) and len(r['phase_seconds'])==5,'phase coverage differs')
        for phase in r['phase_seconds']:finite(phase,'window phase seconds')
        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:
        integer(r['start_window'],'batch start index')
        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(all(w['absolute_encoder_position'] <= w['processed_absolute_encoder_position']
                    <= (startup+end-1)*4 for w in windows[index:end]),'processed encoder clock differs')
        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')
        require(allocated == profile['allocation'], 'rolling 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'] == (source_end_samples(end-1, profile) > case['samples']), 'invalid EOF flag')
        deadline = min(case['samples'],source_end_samples(end-1, profile))/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']
    integer(final['text_bytes'],'text bytes')
    integer(final['visible_tokens'],'visible tokens')
    require(isinstance(final['skipped_counts'],list) and len(final['skipped_counts'])==5,'invalid skip count shape')
    for value in final['skipped_counts']:integer(value,'skipped token count')
    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)
    first = next((r for r in deltas if r['text']), None)
    readable = next((r for r in deltas if any(not c.isspace() and c != '\ufffd' for c in r['text'])), None)
    dependencies=[min(case['samples'],source_end_samples(i,profile))/16000 for i in range(count)]
    def first_details(row):
        if row is None:return None
        i=row['index']
        return dict(index=i,text=row['text'],native_audio_clock_seconds=row['seconds'],
                    source_dependency_seconds=dependencies[i] if i<count else case['seconds'],
                    eof_utf8_flush=row.get('eof_utf8_flush',False))
    report.update(startup_policy=profile['startup_policy'],transcription_delay_ms=expected_delay_ms,
        source_dependency_seconds=dependencies,first_text=first_details(first),
        first_readable_text=first_details(readable),batch_work_rtf=total_work/case['seconds'],
        elapsed_including_input_wait_rtf=elapsed/case['seconds'],
        startup_work_seconds=batches[0]['work_seconds'],
        eof_drain_batches=sum(r['input_eof'] for r in batches),
        eof_drain_work_seconds=sum(r['work_seconds'] for r in batches if r['input_eof']),
        partial_batches=sum(r['end_window_exclusive']-r['start_window'] != batch for r in batches[1:]),
        automatic_acceptance_performed=False,production_accuracy_approved=False,
        latency_scope='First output measured from native audio-clock origin and separately source dependency. Not microphone, speech-onset or aligned word latency.')
    if case.get('reference') is not None:
        require(isinstance(case['reference'],str),'reference must be text')
        report['score']=score(case['reference'],text,case['language'])
    return report

def output_paths(output):
    output=Path(os.path.abspath(output));output=output.parent.resolve()/output.name
    require(output.suffix=='.json','output must end in .json')
    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'),
               observed=output.with_suffix('.observed.jsonl'))
    for p in paths.values():require(not p.exists() and not p.is_symlink(),f'output already exists: {p}')
    return paths


def capture_process(command, source, samples, paths, *, timeout_seconds, max_events, ready_timeout=60,
                    ready_kind='native_stream_ready'):
    """Bounded process group and flushed observations; no model-specific parsing."""
    require(isinstance(ready_kind,str) and bool(ready_kind),'invalid readiness event kind')
    cancel=threading.Event();ready=threading.Event()
    errors=[];rows=[];observations=[];feed_records=[];ready_received=[]
    child=consumer=feeder=None
    clock={}
    def kill_group():
        if child is not None:
            # Clean the whole owned group even if the time wrapper exited first.
            try:os.killpg(child.pid,signal.SIGTERM)
            except ProcessLookupError:pass
            try:child.wait(timeout=5)
            except subprocess.TimeoutExpired:
                try:os.killpg(child.pid,signal.SIGKILL)
                except ProcessLookupError:pass
                child.wait(timeout=5)
    try:
        with contextlib.ExitStack() as stack:
            logs={k:stack.enter_context(paths[k].open('x')) for k in
                  ('raw','timing','errors','feeder','observed')}
            clock['process_started_monotonic']=time.monotonic()
            child=subprocess.Popen(['/usr/bin/time','-l','-o',str(paths['timing']),*command],
                cwd=ROOT,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=logs['errors'],
                start_new_session=True,bufsize=0)
            def consume():
                try:
                    while line:=child.stdout.readline(1024*1024+1):
                        now=time.monotonic()  # Complete line receipt, before parsing/log I/O.
                        require(len(line)<=1024*1024 and line.endswith(b'\n'),'oversized/unflushed native event')
                        logs['raw'].write(line.decode('utf-8'));logs['raw'].flush()
                        row=json.loads(line)
                        require(isinstance(row,dict),'native event is not an object')
                        observation=dict(event_index=len(rows),kind=row.get('kind'),index=row.get('index'),
                                         received_monotonic_seconds=now,
                                         since_process_start_seconds=now-clock['process_started_monotonic'],
                                         received_seconds_since_feeder_start=(now-clock['feeder_started_monotonic'])
                                         if 'feeder_started_monotonic' in clock else None)
                        rows.append(row);observations.append(observation)
                        logs['observed'].write(json.dumps(observation)+'\n');logs['observed'].flush()
                        require(len(rows)<=max_events,'native event bound exceeded')
                        if row.get('kind')==ready_kind:
                            ready_received.append(now)
                            require(len(ready_received)==1,'duplicate native ready event')
                            ready.set()
                except Exception as e:
                    errors.append(f'consumer: {type(e).__name__}: {e}');cancel.set()
                finally:ready.set()
            def feed():
                try:
                    if not ready.wait(ready_timeout):raise TimeoutError('native readiness timeout')
                    require(len(ready_received)==1 and not cancel.is_set(),'no unique ready before feeding')
                    require(child.poll() is None,'child ended before feeding')
                    clock['feeder_started_monotonic']=time.monotonic()
                    count=0
                    with Path(source).open('rb') as src:
                        while chunk:=src.read(320*4):
                            require(len(chunk)%4==0,'partial F32 sample')
                            if cancel.is_set():raise RuntimeError('feeder cancelled')
                            count+=len(chunk)//4
                            deadline=count/16000
                            remaining=deadline-(time.monotonic()-clock['feeder_started_monotonic'])
                            if remaining>0 and cancel.wait(remaining):raise RuntimeError('feeder cancelled')
                            before=time.monotonic()-clock['feeder_started_monotonic']
                            data=memoryview(chunk)
                            while data:
                                wrote=os.write(child.stdin.fileno(),data)
                                if wrote<=0:raise OSError('zero-byte pipe write')
                                data=data[wrote:]
                            after=time.monotonic()-clock['feeder_started_monotonic']
                            record=dict(samples=count,deadline_seconds=deadline,before_seconds=before,
                                        after_seconds=after,write_seconds=after-before)
                            logs['feeder'].write(json.dumps(record)+'\n');logs['feeder'].flush()
                            feed_records.append(record)
                    require(count==samples,'feeder sample coverage differs')
                except Exception as e:
                    errors.append(f'feeder: {type(e).__name__}: {e}');cancel.set()
                finally:
                    try:child.stdin.close()
                    except OSError as e:errors.append(f'close stdin: {e!r}')
            consumer=threading.Thread(target=consume,daemon=True)
            feeder=threading.Thread(target=feed,daemon=True)
            consumer.start();feeder.start()
            try:
                deadline=time.monotonic()+timeout_seconds
                while child.poll() is None:
                    if errors:raise ValueError(f'child I/O failure: {errors}')
                    if time.monotonic()>deadline:raise TimeoutError('candidate process timeout')
                    time.sleep(.02)
                consumer.join(timeout=10);feeder.join(timeout=10)
                require(not consumer.is_alive() and not feeder.is_alive(),'child I/O threads did not finish')
                require(child.returncode==0 and not errors,f'child={child.returncode}; I/O={errors}')
                require(len(ready_received)==1,'no unique native ready event')
            finally:
                cancel.set();ready.set();kill_group()
                for thread in (consumer,feeder):
                    if thread is not None:thread.join(timeout=10)
                child.stdout.close()
                if not child.stdin.closed:child.stdin.close()
    finally:
        cancel.set();ready.set();kill_group()
    require(len(feed_records)==(samples+319)//320 and feed_records[-1]['samples']==samples,
            'feeder chunk/sample coverage differs')
    require(all(r['before_seconds']+1e-7>=r['deadline_seconds'] and r['after_seconds']>=r['before_seconds']
                for r in feed_records),'future source samples fed before availability')
    feeder_summary=dict(chunks=len(feed_records),samples=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,ready_received_monotonic=ready_received[0],
        clock_scope='Feeder starts after host receipt of native readiness. Native and observer clocks are separately recorded; no microphone capture clock.')
    return rows,observations,feeder_summary


def main():
    p=argparse.ArgumentParser(description=__doc__)
    p.add_argument('--binary-record',type=Path,required=True)
    p.add_argument('--model',type=Path,required=True)
    p.add_argument('--model-sha256',required=True)
    p.add_argument('--frontend',type=Path,default=ROOT/'artifacts/frontend-primitive.a8m')
    p.add_argument('--tokenizer',type=Path,default=ROOT/'artifacts/native-tokenizer-v1.a8tok')
    p.add_argument('--input-manifest',type=Path,required=True)
    p.add_argument('--case',required=True)
    p.add_argument('--startup',choices=('hf','realtime'),required=True)
    p.add_argument('--delay-ms',type=int,choices=(80,160,240,480),required=True)
    p.add_argument('--first-batch',type=int,choices=(1,),default=1)
    p.add_argument('--batch',type=int,choices=(1,2,4),required=True)
    p.add_argument('--prefix',type=Path)
    p.add_argument('--timeout-seconds',type=float,default=1200)
    p.add_argument('--allow-heldout',action='store_true',help='Explicit final evaluation stage only; default refuses heldout inputs.')
    p.add_argument('--release-candidate-record',type=Path)
    p.add_argument('--release-candidate-record-sha256')
    p.add_argument('--output',type=Path,required=True)
    a=p.parse_args()
    require(platform.system()=='Darwin','candidate timing harness currently requires macOS time -l')
    require(math.isfinite(a.timeout_seconds) and 1<=a.timeout_seconds<=7200,'invalid bounded timeout')
    profile=candidate_profile(a.startup,a.delay_ms,a.batch,a.first_batch)
    require(len(a.model_sha256)==64 and set(a.model_sha256)<=set('0123456789abcdef'),'invalid model SHA256')
    paths=output_paths(a.output)
    release_candidate=release_identity(a.release_candidate_record,a.release_candidate_record_sha256)
    case=load_input_case(a.input_manifest,a.case);require_input_role(case,a.allow_heldout,release_candidate)
    require(case['seconds']<=3600,'candidate harness input exceeds one-hour bounded scope')
    binary_record=json.loads(a.binary_record.read_text());binary=Path(binary_record['binary']).resolve()
    asset_paths=dict(binary=binary,binary_record=a.binary_record,model=a.model,frontend=a.frontend,
        tokenizer=a.tokenizer,input=Path(case['path']),input_manifest=a.input_manifest,script=Path(__file__),
        token_protocol_helpers=ROOT/'scripts/probe_native_stream.py',scoring_memory_helpers=ROOT/'run_native_probe.py',
        candidate_prefix_helper=ROOT/'scripts/native_candidate_prefix.py')
    if a.prefix:asset_paths['prefix']=a.prefix
    if release_candidate:asset_paths['release_candidate_record']=Path(release_candidate['path'])
    assets={k:dict(path=str(v.resolve()),sha256=digest(v),bytes=v.stat().st_size) for k,v in asset_paths.items()}
    require(assets['binary']['sha256']==binary_record['sha256'],'binary record SHA differs')
    require(assets['model']['sha256']==a.model_sha256,'pinned model SHA differs')
    require(assets['input']['sha256']==case['sha256'] and assets['input']['bytes']==case['samples']*4,
            'input PCM hash/size differs')
    with a.model.open('rb') as f:
        require(f.read(8)==b'A8MOD001' and assets['model']['bytes']>=52,'invalid native model container')
        f.seek(-32,2);body_sha=f.read(32).hex()
    prefix_metadata=None
    if a.prefix:
        _,prefix_metadata=inspect_candidate_prefix(a.prefix,case['language'],dict(frontend=assets['frontend'],
            runtime_binary=assets['binary'],model_stream=assets['model']),profile)
    table=token_table(a.tokenizer)
    command=[str(binary),str(a.model.resolve()),str(a.frontend.resolve()),str(a.tokenizer.resolve()),'-',case['language'],
        '--workers','4','--tasks','1','--worker-qos','user-initiated','--trace-windows',
        '--startup',a.startup,'--delay-ms',str(a.delay_ms),'--first-batch','1','--batch',str(a.batch)]
    if a.prefix:command+=['--prefix-in',str(a.prefix.resolve())]
    report=dict(status='running',started_at_unix=time.time(),command=command,assets=assets,input_case=case,
        requested_profile=profile,batch_windows=a.batch,prefix_metadata=prefix_metadata,
        release_candidate=release_candidate,
        timeout_seconds=a.timeout_seconds,automatic_acceptance_performed=False,production_accuracy_approved=False,
        scope='Candidate continuous stdin probe; actual20ms feeder pacing and flushed visible-text observation. No model selection or automatic quality/latency acceptance. Not microphone or aligned-content latency.')
    paths['report'].parent.mkdir(parents=True,exist_ok=True)
    with paths['report'].open('x') as f:json.dump(report,f,ensure_ascii=False,indent=2)
    error=None
    try:
        rows,observations,feeding=capture_process(command,case['path'],case['samples'],paths,
            timeout_seconds=a.timeout_seconds,max_events=emission_count(case['samples'],profile)*3+200)
        report.update(validate(rows,case,table=table,profile=profile,batch=a.batch,
                               model_body_sha256=body_sha,prefix_metadata=prefix_metadata))
        report['feeder']=feeding
        for label in ('first_text','first_readable_text'):
            if report[label] is not None:
                item=report[label]
                event=next(o for r,o in zip(rows,observations) if r['kind']=='native_text_delta'
                           and r['index']==item['index'] and r['text']==item['text'])
                item['host_observed_seconds_since_feeder_start']=event['received_monotonic_seconds']-feeding['feeder_started_monotonic']
        report['memory']=parse_time(paths['timing'].read_text(),'Darwin')
        report['assets_unchanged']=all(digest(v['path'])==v['sha256'] for v in assets.values())
        require(report['assets_unchanged'],'candidate asset changed during run')
        report['status']='complete'
    except Exception as e:
        error=e;report.update(status='failed',error=f'{type(e).__name__}: {e}')
    finally:
        report['finished_at_unix']=time.time()
        report['logs']={k:dict(path=str(v),sha256=digest(v)) for k,v in paths.items() if k!='report' and v.exists()}
        paths['report'].write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
    if error:raise error
    print(json.dumps({k:report.get(k) for k in ('status','text','score','first_readable_text','batch_work_rtf',
        'source_active_p95_work_seconds_per_window','max_queue_seconds','memory')},ensure_ascii=False),flush=True)


if __name__=='__main__':main()