from ops import *
import ast,signal,sys,yaml
label=sys.argv[1];assert label in ('pose','mixed','mixed-resume','mixed-compact')
mode='pose' if label=='pose' else 'mixed'
levels=range(8,17) if label in ('mixed-resume','mixed-compact') else range(1,17)
BASE=HERE;HERE=BASE/label;HERE.mkdir(mode=0o700)
def save(name,data,private=False):
    with ((PRIVATE if private else HERE)/name).open('x') as f:json.dump(data,f,indent=2)
ORDER=[3,5,6,8,9,10,11,12,13,14,15,16,17,18,19,20]
KEYS=('pose',) if mode=='pose' else ('pose','fire','crowd')
PROBE='/opt/contextra/tools/provider_scale_probe.py'
original=yaml.safe_load((PRIVATE/'runtime-original.yaml').read_text())['added']
specs={c:next(s for s in original if str(s.get('camera_id'))==str(c)) for c in ORDER}
sids={c:specs[c]['stream_id'] for c in ORDER};reverse={v:k for k,v in sids.items()}
FIELDS=('intrusion','loitering','fall','fight','fire','crowd','tracking')
saved={c:api('/api/v1/streams/'+sids[c]+'/settings')['configured'] for c in ORDER}
save('settings-'+label+'.json',saved,private=True)
starts={};touched=[];launched=[];begin=time.monotonic();initial_oom=None;control_pid=None
tree=ast.parse(Path('/tmp/codex/ltb1021/2026-09-26_provider-scale_r01/trial.py').read_text())
names={'event','interrupt','patch_settings','sample','model_config','frame_probe'}
exec(compile(ast.Module(body=[n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name in names],type_ignores=[]),'<reused-probes>','exec'))
def rows():
    out={}
    for c in launched:
        started=time.monotonic()
        try:r=api('/api/v1/streams/'+sids[c])
        except Exception as error:
            event('status_query_error',camera=c,type=type(error).__name__,seconds=time.monotonic()-started)
            raise
        elapsed=time.monotonic()-started
        if elapsed>1:event('slow_status_query',camera=c,seconds=elapsed)
        out[str(c)]={k:r.get(k) for k in ('state','run_id','run_generation','last_error','cleanup_error','cleanup_pending','degraded','scenarios','publisher_metrics')}
        out[str(c)]['providers']={p:{k:v.get(k) for k in ('state','child_pid','last_error','cleanup_error','last_cleanup','metadata_path','publisher_metrics')}
            for p,v in (r.get('provider_processes') or {'pose':r}).items()}
    return out
def start_channel(c):
    basic_safety(rows());launched.append(c)
    sent=time.monotonic();wall=time.time_ns()//1000000
    ack=request('/api/v1/streams/'+sids[c]+'/start',{},timeout=20)
    starts[c]=dict(sent=sent,wall_ms=wall,ack_at=time.monotonic(),request_state=ack.get('state'))
    event('start_ack',camera=c,**starts[c]);deadline=time.monotonic()+90
    while True:
        current=rows();persist(current,'startup-'+str(c));basic_safety(current);r=current[str(c)]
        ready=len(r['providers'])==len(KEYS) and all(v.get('metadata_path',{}).get('providers',{}).get(p,{}).get('responses',0)>=2 for p,v in r['providers'].items())
        if ready:
            starts[c]['ready_observed_at']=time.monotonic();break
        if time.monotonic()>deadline:raise RuntimeError('NO_PROVIDER_RESPONSE_'+str(c))
        time.sleep(1)
def basic_safety(current):
    if time.monotonic()-begin>2100:raise RuntimeError('TRIAL_TOTAL_BOUND')
    if shutil.disk_usage('/').free<15*1024**3:raise RuntimeError('ROOT_FREE_BOUND')
    available=int(next(l for l in Path('/proc/meminfo').read_text().splitlines() if l.startswith('MemAvailable:')).split()[1])*1024
    if available<2*1024**3:raise RuntimeError('HOST_RAM_FREE_BOUND')
    for c,r in current.items():
        if r['last_error'] or r['cleanup_error'] or r['degraded']:raise RuntimeError('CHANNEL_ERROR_'+c)
        for p,v in r['providers'].items():
            if v['last_error'] or v['cleanup_error'] or v['state']=='error':raise RuntimeError('PROVIDER_ERROR_'+c+'_'+p)
def persist(current,where):
    with (HERE/'startup.jsonl').open('a') as f:
        f.write(json.dumps(dict(t=time.monotonic(),wall_ms=time.time_ns()//1000000,where=where,channels=current))+'\n')
def protected_paths(current):
    for c,r in current.items():
        assert len(r['providers'])==len(KEYS),(c,'PROVIDER_COUNT')
        for p,v in r['providers'].items():
            path=v['metadata_path'];art=path.get('artifacts') or {}
            assert path['execution_path']=='provider-linear-trial' and path['source_latest_trial']
            assert not path['freshness_enabled'] and path['queue_max_age_s']==1
            assert art.get('mode')=='test_no_capture' and not art['delivery_enabled']
            assert not art['encoded_cache_enabled'] and not art['training_enabled']
            assert not art['queued'] and not art['pending']
            d=path['performance']['pipeline_diagnostic']['providers'][p]
            if label=='mixed-compact':assert d.get('input_timeline_limit')==2
            assert d['collecting'] and d['input_timeline'] and any(x['input'].get('latest_applied')==1 for x in d['input_timeline']), 'TIMING_OR_LATEST_UNAVAILABLE'
signal.signal(signal.SIGTERM,interrupt);signal.signal(signal.SIGINT,interrupt)
assert not inspect('contextra-ai-gateway-1')['State']['Running']
result=dict(mode=mode,stages=[],starts=starts,errors=[],cleanup=None)
try:
    configs=model_config();save('model-configs.json',configs)
    capture_before=ledger();training_before=training();save('capture-before.json',capture_before)
    save('training-before.json',training_before)
    conditions=[]
    for c in ORDER:
        desired={k:dict(enabled=True) for k in ('intrusion','loitering','fall','fight','tracking')}
        desired.update(fire=dict(enabled=mode=='mixed',scenario=dict(enabled=mode=='mixed')),crowd=dict(enabled=mode=='mixed'))
        touched.append(c);v=patch_settings(c,desired)['configured']
        assert [p['key'] for p in v['execution_plan']['providers']]==list(KEYS)
        assert v['analysisFpsTarget']==10
        conditions.append(dict(camera=c,source_sha256=hashlib.sha256(specs[c]['uri'].encode()).hexdigest(),
            source_options_sha256=hashlib.sha256(json.dumps(specs[c].get('source_options',{}),sort_keys=True).encode()).hexdigest(),
            original_revision=saved[c]['revision'],trial_revision=v['revision'],scenarios={k:v[k] for k in FIELDS},pose=v['pose']))
    save('conditions.json',dict(order=ORDER,levels=list(levels),duration_s=30,warmup_s=5,conditions=conditions,
         status_query='existing per-stream GET; unchanged5s timeout',continuation_from='mixed' if label in ('mixed-resume','mixed-compact') else None,
         timeline_limit=2 if label=='mixed-compact' else 8,
         limits=dict(total_s=2100,native_timing_s=2400,root_free_gib=15,host_free_ram_gib=2)))
    for level in levels:
        stage=dict(level=level,index=level-1,seconds=30,samples=[],status='RUNNING')
        result['stages'].append(stage);event('stage_begin',level=level)
        for c in ORDER[:level]:
            if c not in launched:start_channel(c)
        time.sleep(5);basic_safety(rows());start=time.monotonic()
        stage.update(started_wall_ms=time.time_ns()//1000000,started_mono=start)
        for i in range(7):
            time.sleep(max(0,start+i*5-time.monotonic()))
            value=sample();basic_safety(value['channels']);protected_paths(value['channels'])
            stage['samples'].append(value)
            with (HERE/'samples.jsonl').open('a') as f:f.write(json.dumps(dict(stage=level-1,level=level,**value))+'\n')
            event('sample',level=level,index=i)
        stage['frames']=frame_probe(ORDER[:level]);assert configs==model_config()
        capture_now=ledger();stage['capture_ledger']=capture_now
        assert capture_now['count']==capture_before['count'],'NEW_EVENT_LEDGER_DURING_C'
        stage['status']='PASS';stage['finished_wall_ms']=time.time_ns()//1000000
        save('stage-'+str(level).zfill(2)+'.json',stage)
        print(json.dumps(dict(mode=mode,level=level,status='PASS',seconds=30)),flush=True)
    save('capture-after.json',ledger());save('training-after.json',training())
except Exception as e:
    import traceback
    result['errors'].append(dict(type=type(e).__name__,reason=str(e)[:160],
        call_path=[dict(function=x.name,line=x.lineno,file=Path(x.filename).name) for x in traceback.extract_tb(e.__traceback__)]))
    event('trial_error',**result['errors'][-1])
finally:
    for c in launched:
        try:request('/api/v1/streams/'+sids[c]+'/stop',{},timeout=20)
        except Exception as e:result['errors'].append(dict(type=type(e).__name__,reason='STOP_REQUEST_'+str(c)))
    deadline=time.monotonic()+120
    while True:
        r=rows();reaped=all(not v['cleanup_pending'] and all(not p.get('child_pid') for p in v['providers'].values()) for v in r.values())
        if reaped or time.monotonic()>deadline:break
        time.sleep(1)
    result['cleanup']=dict(reaped=reaped,states=r)
    if reaped:
        for c in touched:
            try:
                if c in launched and r[str(c)]['state']!='stopped':raise RuntimeError('RESTORE_DEFERRED_ERROR_STATE')
                changes={k:saved[c][k] for k in FIELDS}
                if 'scenario' not in changes['fire']:changes['fire']=dict(changes['fire'],scenario={'enabled':False})
                patch_settings(c,changes)
            except Exception as e:result['errors'].append(dict(type=type(e).__name__,reason='RESTORE_SETTINGS_'+str(c)))
    save('trial-result.json',result);print(json.dumps(dict(mode=mode,errors=result['errors'],reaped=reaped)),flush=True)
if result['errors'] or not reaped:raise SystemExit(1)