import sys,json,time,subprocess,hashlib
from pathlib import Path
sys.path.insert(0,'/tmp/codex/ltb1021/2026-09-26_linear-scale_r01')
import ops
ROOT=Path(__file__).parent;ROOT.chmod(0o700)
def save(name,value):
    with (ROOT/name).open('x') as f:json.dump(value,f,indent=2)
def current():
    rows=ops.api('/api/v1/streams')['streams']
    active=[r for r in rows if r['state']!='stopped']
    assert len(active)==1
    return active[0]
before=current();run=before['run_id'];sid=before['stream_id']
def artifact():
    sql="""SELECT coalesce(json_agg(t),'[]'::json) FROM (
    SELECT payload->>'scenario' AS scenario,artifact_state,delivery_state,http_status,
    CASE WHEN error ~ '^[A-Z0-9_:-]+$' THEN left(error,80) WHEN error IS NULL THEN NULL ELSE 'OTHER' END AS error,
    count(*) AS count,min(created_at) AS first_created,max(created_at) AS last_created,
    max(extract(epoch from now())-created_at) FILTER (WHERE artifact_state IN ('CAPTURING','CAPTURE_READY','RENDERING')) AS oldest_pending_seconds,
    avg(updated_at-confirmed_at) FILTER (WHERE delivery_state='SENT') AS confirmed_to_last_update_mean_seconds,
    max(updated_at-confirmed_at) FILTER (WHERE delivery_state='SENT') AS confirmed_to_last_update_max_seconds
    FROM contextra_event_artifacts WHERE payload->>'cameraId'='3' AND payload->>'runId'='RUN'
    GROUP BY 1,2,3,4,5) t;""".replace('RUN',run)
    return json.loads(ops.cmd(['docker','exec','-i','contextra-postgresql-1','sh','-c','exec psql -X -qAt -U "$POSTGRES_USER" -d "$POSTGRES_DB"'],input=sql))
a0=artifact();start=time.monotonic()
code=r'''
import sys,json,time,os,hashlib,statistics
import redis
from urllib.parse import urlsplit
auth=json.load(sys.stdin);u=urlsplit(os.environ['CONTEXTRA_METADATA_REDIS_URL'])
c=redis.Redis(host=u.hostname,port=u.port or 6379,db=int(u.path.strip('/') or 0),username=auth['username'],password=auth['password'],decode_responses=True,socket_connect_timeout=2,socket_timeout=3)
key=os.environ.get('CONTEXTRA_METADATA_NAMESPACE','contextra:metadata:v1')+':camera:'+hashlib.sha256(b'3').hexdigest()+':frames'
latest=c.xrevrange(key,count=1);cursor=latest[0][0] if latest else '0-0'
begin=time.monotonic();rows={};other=0;cross_provider_seq_nonincrease=0;previous_seq=None
while time.monotonic()-begin<30:
 for _,entries in c.xread({key:cursor},count=1000,block=300):
  for eid,fields in entries:
   cursor=eid;s=json.loads(fields['json'])
   if s.get('runId')!='RUN':other+=1;continue
   p=s.get('providerKey');seq=s['frameSeq'];pts=s.get('ptsNs');ms=int(eid.split('-')[0]);objects=s.get('objects',[])
   r=rows.setdefault(p,dict(count=0,first_seq=seq,last_seq=seq,first_ms=ms,last_ms=ms,missing_seq=0,duplicate_or_backward=0,pts_backward=0,last_pts=None,objects=0,visible_objects=0,visible_frames=0,empty_frames=0,event_count=0,models=[],required=[],interval_ms=[],object_keys=[]))
   if r['count']:
    r['missing_seq']+=max(0,seq-r['last_seq']-1);r['duplicate_or_backward']+=int(seq<=r['last_seq']);r['interval_ms'].append(ms-r['last_ms'])
    if pts is not None and r['last_pts'] is not None:r['pts_backward']+=int(pts<r['last_pts'])
   if previous_seq is not None:cross_provider_seq_nonincrease+=int(seq<=previous_seq)
   previous_seq=seq;r['count']+=1;r['last_seq']=seq;r['last_ms']=ms;r['last_pts']=pts
   visible=sum(bool(o.get('visible')) for o in objects)
   r['objects']+=len(objects);r['visible_objects']+=visible;r['visible_frames']+=int(visible>0);r['empty_frames']+=int(not objects);r['event_count']+=len(s.get('events',[]))
   if s.get('providerModel') not in r['models']:r['models'].append(s.get('providerModel'))
   if s.get('requiredProviders') not in r['required']:r['required'].append(s.get('requiredProviders'))
   if objects and not r['object_keys']:r['object_keys']=sorted(objects[0])
elapsed=time.monotonic()-begin
for r in rows.values():
 intervals=r.pop('interval_ms');r['fps']=r['count']/elapsed;r['interval_mean_ms']=statistics.mean(intervals) if intervals else None;r['interval_max_ms']=max(intervals) if intervals else None
 r['interval_p95_ms']=sorted(intervals)[min(len(intervals)-1,int(len(intervals)*.95))] if intervals else None
c.close();print(json.dumps(dict(seconds=elapsed,providers=rows,other_runs=other,cross_provider_seq_nonincrease=cross_provider_seq_nonincrease)))
'''.replace('RUN',run)
auth=Path('/workspace/contextra-runtime-config/metadata-active/reader/reader-credentials.json')
assert auth.stat().st_mode&0o077==0
redis=json.loads(ops.cmd(['docker','exec','-i','contextra-analytics-1','python3','-B','-c',code],input=auth.read_text(),timeout=40))
after=current();assert after['run_id']==run and after['stream_id']==sid
a1=artifact();dt=time.monotonic()-start
def compact(row):
    return {k:row.get(k) for k in ('state','run_id','run_generation','last_error','degraded','provider_processes','publisher_metrics')}
save('observation.json',dict(seconds=dt,start_wall_ms=time.time_ns()//1000000-int(dt*1000),end_wall_ms=time.time_ns()//1000000,before=compact(before),after=compact(after),redis=redis,artifacts_before=a0,artifacts_after=a1))
print(json.dumps(dict(seconds=dt,redis=redis,artifacts_before=a0,artifacts_after=a1),indent=2))