from ops import *
import collections
targets=json.loads((HERE/'baseline.json').read_text())['targets']
save('trial-intent.json',dict(cameras=[t['camera'] for t in targets],seconds=30,settings='unchanged'))
def cpu_snapshot():
    rows=cmd(['docker','top','contextra-analytics-1','-eo','pid,comm']).splitlines()[1:]
    statuses={r['child_pid']:next(t['camera'] for t in targets if t['stream_id']==r['stream_id']) for r in api('/api/v1/streams')['streams'] if r['state']=='running'}
    out={}
    for line in rows:
        pid,comm=line.split()
        if comm not in ('python3','tritonserver'):continue
        root=Path('/proc')/pid
        try:ns=int(next(x for x in (root/'status').read_text().splitlines() if x.startswith('NSpid:')).split()[-1])
        except (OSError,StopIteration):continue
        label='Triton' if comm=='tritonserver' else str(statuses.get(ns,'parent'))
        for t in (root/'task').glob('*'):
            try:out[pid+':'+t.name]=dict(label=label,name=(t/'comm').read_text().strip(),values=[int(x) for x in (t/'schedstat').read_text().split()])
            except OSError:pass
    return out
out={};started=[]
try:
    for t in targets:
        sid=t['stream_id'];started.append(sid);request('/api/v1/streams/'+sid+'/start',{})
    deadline=time.monotonic()+45
    while True:
        rows=[api('/api/v1/streams/'+t['stream_id']) for t in targets]
        if any(r['state']=='error' for r in rows):raise RuntimeError('CHANNEL_ERROR')
        if all(r['state']=='running' and r.get('metadata_path',{}).get('providers') and all(v['responses']>=5 for v in r['metadata_path']['providers'].values()) for r in rows):break
        if time.monotonic()>deadline:raise RuntimeError('FIRST_RESULT_TIMEOUT')
        time.sleep(1)
    c0=cpu_snapshot();start=time.monotonic()
    p=Path('/tmp/codex/ltb1021/2026-09-23_pr13-load30_r01/observe.py')
    exec(compile(p.read_text(),str(p),'exec'),{'__file__':str(HERE/'observe.py'),'__name__':'observer'})
    elapsed=time.monotonic()-start;c1=cpu_snapshot();aggregates={};threads=[]
    for k,b in c1.items():
        if k not in c0:continue
        a=c0[k];run=(b['values'][0]-a['values'][0])/1e9;wait=(b['values'][1]-a['values'][1])/1e9
        o=aggregates.setdefault(b['label'],dict(cpu_s=0.,runnable_wait_s=0.,threads=0));o['cpu_s']+=run;o['runnable_wait_s']+=wait;o['threads']+=1
        threads.append(dict(label=b['label'],name=b['name'],cpu_s=run,runnable_wait_s=wait))
    save('cpu.json',dict(seconds=elapsed,by_channel=aggregates,top_wait=sorted(threads,key=lambda x:x['runnable_wait_s'],reverse=True)[:20]))
    out['observed']=True
except Exception as e:
    out['error']=type(e).__name__+':'+str(e)[:100]
finally:
    for sid in started:request('/api/v1/streams/'+sid+'/stop',{},timeout=90)
    deadline=time.monotonic()+25
    while True:
        rows=api('/api/v1/streams')['streams']
        if all(r['state']=='stopped' and not r.get('child_pid') and not r.get('cleanup_pending') for r in rows):break
        if time.monotonic()>deadline:raise RuntimeError('STOP_REAP_FAILED')
        time.sleep(1)
    before=json.loads((PRIVATE/'settings-before.json').read_text())
    out['settings_unchanged']=all(api('/api/v1/streams/'+sid+'/settings')['configured']==cfg for sid,cfg in before.items())
    out['stopped']=len(rows)
    save('trial-result.json',out)
    print(json.dumps(out),flush=True)
if out.get('error'):raise SystemExit(1)