import importlib.util
import json
import os
import signal
import subprocess
import sys
import time
import traceback
from pathlib import Path
import hashlib
import csv
import yaml

os.umask(0o077)
R=Path(__file__).parent
spec=importlib.util.spec_from_file_location('ops','/tmp/codex/ltb1021/2026-09-15_contextra-redis-final_r01/ops.py')
ops=importlib.util.module_from_spec(spec);spec.loader.exec_module(ops)
api,query,REPO=ops.m.api,ops.m.query,ops.REPO
sys.path.insert(0,str(REPO/'tools'))
from rtsp_test_manager import Manager
TARGET='rtsp-test-01'
P='/api/v1/streams/'+TARGET

def save(name,value):
    (R/name).write_text(json.dumps(value,ensure_ascii=False,indent=2))

def sender():
    return next(x for x in Manager().status() if x['name']=='sample01')

def mem():
    values=dict(line.split(':',1) for line in Path('/proc/meminfo').read_text().splitlines())
    return int(values['MemAvailable'].split()[0])*1024

def resource():
    info=ops.inspect('contextra-analytics-1');pid=info['State']['Pid']
    status=Path('/proc/'+str(pid)+'/status').read_text()
    rss=next(int(l.split()[1])*1024 for l in status.splitlines() if l.startswith('VmRSS:'))
    proc=Path('/proc/'+str(pid)+'/stat').read_text().split(') ',1)[1].split()
    return dict(wall=time.monotonic(),pid=pid,rss_bytes=rss,process_cpu_ticks=int(proc[11])+int(proc[12]),mem_available=mem())

def gpu():
    raw=subprocess.check_output(['nvidia-smi','--query-gpu=memory.used,memory.total,utilization.gpu','--format=csv,noheader,nounits'],text=True,timeout=5)
    return [list(map(int,l.split(','))) for l in raw.splitlines()]

def guard_resource():
    x=resource()
    assert x['rss_bytes']<12*1024**3 and x['mem_available']>8*1024**3,'RAM_LIMIT'
    return x

def xinfo(key):
    x=query('reader','XINFO','STREAM',key)
    return x if isinstance(x,dict) else dict(zip(x[::2],x[1::2]))

def gate():
    assert not (R/'plan.json').exists(),'ALREADY_PLANNED'
    head=subprocess.check_output(['git','rev-parse','HEAD'],cwd=REPO,text=True).strip()
    assert head=='b830b702ccdd4ee76842851bb9fc559e2179ff80'
    assert not subprocess.check_output(['git','status','--porcelain'],cwd=REPO,text=True).strip()
    streams=api('/api/v1/streams')['streams'];target=next(x for x in streams if x['stream_id']==TARGET)
    assert target['state'] in ('stopped','running')
    conf=api(P+'/settings')['configured']
    reg=yaml.safe_load(Path('/data/contextra/runtime-state/streams_runtime.yaml').read_text())
    cfg=next(x for x in reg['added'] if x['stream_id']==TARGET)
    send=sender();assert send['state']=='STREAMING' and send['mode']=='continuous' and send['anchor']
    assert send['media']['lastPacketAgeSeconds']<2
    assert send['uri']==cfg['uri']
    assert cfg['store'] is False and cfg['active_learning'] is False
    assert not cfg.get('capture',{}).get('enabled',False)
    assert conf['execution_plan']['model']=='RFDETRMedium' and conf['analysisFpsTarget']==10
    assert conf['pose']['enabled'] is False and conf['fire']['enabled'] is False
    intr=conf['intrusion'];assert intr['enabled'] and intr['confidence']==.5 and intr['dwell_seconds']==.5
    assert intr['roi']==[[.35,.3],[.85,.3],[.85,1],[.35,1]]
    assert conf['display_categories']==['person'] and not conf['debug_all_candidates']
    probe=json.loads(subprocess.check_output(['ffprobe','-v','error','-select_streams','v:0','-show_entries','stream=codec_name,width,height,r_frame_rate','-of','json',send['video']],timeout=10))
    assert probe['streams'][0]['codec_name']=='h264'
    image=ops.inspect('contextra-analytics-1');assert image['State']['Health']['Status']=='healthy'
    env=dict(v.split('=',1) for v in image['Config']['Env'] if '=' in v)
    assert not env.get('CONTEXTRA_METADATA_VISUAL_DIR') and not env.get('CONTEXTRA_START_DIAGNOSTICS_DIR')
    assert env['CONTEXTRA_METADATA_MAX_MESSAGE_BYTES']=='10485760' and env['CONTEXTRA_METADATA_MAX_BYTES']=='268435456'
    assert query('admin','CONFIG','GET','maxmemory','maxmemory-policy')=={'maxmemory':'8589934592','maxmemory-policy':'allkeys-lru'}
    res=guard_resource();gp=gpu();assert all(x[0]<16000 for x in gp),'GPU_BUSY_CAP'
    ns='contextra:metadata:v1';key=ns+':camera:'+hashlib.sha256(cfg['source_id'].encode()).hexdigest()+':frames'
    files=[ops.FILES['acl'],ops.FILES['config'],ops.FILES['compose'],ops.FILES['env'],ops.FILES['access'],ops.FILES['firewall']]
    plan=dict(started=time.time(),head=head,target_state=target,configured=conf,registration=cfg,sender=send,probe=probe,image_id=image['Id'],image_digest=image['Image'],others=ops.inventory(),redis_id=ops.inspect('contextra-redis-1')['Id'],other_streams=[x for x in streams if x['stream_id']!=TARGET],key=key,resource=res,gpu=gp,config_hashes={str(p):ops.g.sha(p) for p in files},publisher=api('/api/v1/metadata/status'),caps=dict(helper_seconds=150,first_frame_seconds=40,warmup_seconds=5,measure_seconds=20,stored_frames=90,observed_frames=600,rss_bytes=12*1024**3,min_available_ram_bytes=8*1024**3,gpu_used_mib=16000,diagnostic_seconds=30,evidence_bytes=16*1024**2),rollback='On failed new run Stop target only, confirm cleanup, restore only changed settings via CAS; no sender/other-stream stop. On success retain running AI/sender, disable metrics only.',keep_running_on_success=True)
    save('plan.json',plan);save('ledger.json',dict(start=0,stop=0,patch=0,metrics_on=0,metrics_off=0,sender_start=0))
    print(json.dumps(dict(gate='PASS',state=target['state'],anchor_matches=conf.get('frame_time_anchor')==send['anchor']['frame_time_anchor'],store=False,al=False,capture=False,codec='h264',model='RFDETRMedium')))

def run():
    plan=json.loads((R/'plan.json').read_text());caps=plan['caps'];ledger=json.loads((R/'ledger.json').read_text())
    assert not (R/'start-attempt.json').exists(),'ALREADY_ATTEMPTED'
    state=dict(stage='PREPARED',samples=[]);frames={};event_ids=set();seen=set();red=0;metrics_on=False;started=False;success=False
    def alarm(*_):raise TimeoutError('HELPER_CAP')
    signal.signal(signal.SIGALRM,alarm);signal.alarm(caps['helper_seconds'])
    try:
        conf=api(P+'/settings')['configured'];assert conf==plan['configured'],'SETTINGS_DRIFT'
        send=sender();assert send['executionId']==plan['sender']['executionId']
        if conf.get('frame_time_anchor')!=send['anchor']['frame_time_anchor']:
            assert plan['target_state']['state']=='stopped','RUNNING_ANCHOR_DRIFT'
            ledger['patch']=1;save('ledger.json',ledger)
            patched=api(P+'/settings','PATCH',dict(expected_revision=conf['revision'],frame_time_anchor=send['anchor']['frame_time_anchor']))
            save('settings-after.json',patched);conf=patched['configured']
        save('applied-anchor.json',dict(sender=send,settings_revision=conf['revision'],frame_time_anchor=conf['frame_time_anchor']))
        current=api(P)
        save('start-attempt.json',dict(at=time.time(),state=current['state']))
        if current['state']=='stopped':
            ledger['start']=1;save('ledger.json',ledger);started=True
            state['start_response']=api(P+'/start','POST')
        else:assert current['state']=='running'
        deadline=time.monotonic()+caps['first_frame_seconds'];boot=time.monotonic()
        while time.monotonic()<deadline:
            current=api(P)
            assert not current.get('last_error') and not current.get('cleanup_error'),'RUNTIME_ERROR'
            if current['state']=='running' and current.get('metadata_path',{}).get('processed',0)>0:break
            time.sleep(.4)
        else:raise TimeoutError('NO_FIRST_FRAME')
        state['first_frame_wait_seconds']=time.monotonic()-boot
        state['run_id']=current['run_id'];state['generation']=current['run_generation']
        time.sleep(caps['warmup_seconds'])
        ledger['metrics_on']=1;save('ledger.json',ledger)
        api(P+'/metrics','POST',dict(run_generation=state['generation'],enabled=True,seconds=caps['diagnostic_seconds']));metrics_on=True
        before=api(P);x0=xinfo(plan['key']);t0=time.monotonic();r0=guard_resource()
        previous=None;latestid='0-0';stop=t0+caps['measure_seconds'];lengths=[];ttls=[]
        while time.monotonic()<stop:
            entries=query('reader','XRANGE',plan['key'],'('+latestid,'+','COUNT','30') or []
            for eid,fields in entries:
                latestid=eid
                if isinstance(fields,list):fields=dict(zip(fields[::2],fields[1::2]))
                frame=json.loads(fields['json'])
                assert frame['runId']==state['run_id'] and frame['runGeneration']==state['generation']
                assert frame['sourceId']==plan['registration']['source_id'] and frame['cameraId']==frame['sourceId']
                assert frame['serverId']==plan['registration']['server_id']
                seq=frame['frameSeq'];assert seq not in seen;seen.add(seq)
                assert len(seen)<=caps['observed_frames']
                ft=frame['frameTime'];assert ft['rtpTimestamp'] is not None and ft['rtpClockRate']==90000
                assert ft['startWallTimeMs']==conf['frame_time_anchor']['startWallTimeMs']
                assert ft['syncStatus']=='ESTIMATED'
                assert ft['detectedAtWallMs']==ft['startWallTimeMs']+ft['rtpTimestamp']*1000//ft['rtpClockRate']
                assert frame['coordinateSpace']=='SOURCE_FRAME' and frame['normalized'] is True and frame['format']=='XYXY'
                for obj in frame['objects']:
                    assert obj['modelKey']=='RFDETRMedium' and obj['classId']==0
                    assert len(obj['bbox'])==4 and 0<=obj['confidence']<=1
                    for sr in obj.get('scenarioResults',[]):
                        assert sr['scenario']=='INTRUSION'
                        if sr['state']=='INTRUSION':assert obj['color']=='#FF3030';red+=1
                for event in frame.get('events',[]):
                    assert event['frameSeq']==seq and event['runId']==state['run_id']
                    assert event['deliveryStatus']=='NOT_SUBMITTED' and event['clipUri'] is None and event['thumbnailUri'] is None
                    event_ids.add(event['eventId'])
                if len(frames)<30 or (seq%4==0 and len(frames)<caps['stored_frames']):frames[seq]=frame
            current=api(P);assert current['state']=='running' and not current.get('last_error')
            lengths.append(query('reader','XLEN',plan['key']));ttls.append(query('reader','PTTL',plan['key']))
            assert 0<lengths[-1]<=30 and 0<ttls[-1]<=10000
            state['samples'].append(dict(at=time.monotonic()-t0,processed=current['metadata_path']['processed'],queued=current['metadata_path']['queued'],last_seq=max(seen) if seen else None))
            guard_resource();time.sleep(.5)
        x1=xinfo(plan['key']);t1=time.monotonic();after=api(P);r1=guard_resource();gp=gpu()
        assert all(x[0]<caps['gpu_used_mib'] for x in gp),'GPU_LIMIT'
        assert seen,'NO_REDIS_DATA'
        result_fps=(after['metadata_path']['processed']-before['metadata_path']['processed'])/(t1-t0)
        redis_fps=(x1['entries-added']-x0['entries-added'])/(t1-t0)
        assert result_fps>=9 and redis_fps>=9,'THROUGHPUT_BELOW_BOUNDED_ACCEPTANCE'
        assert max(x['queued'] for x in state['samples'])<=4 and state['samples'][-1]['queued']<=1,'QUEUE_BUILDUP'
        pub=api('/api/v1/metadata/status')
        assert pub['metrics'].get('send_error',0)==plan['publisher']['metrics'].get('send_error',0),'PUBLISH_ERROR'
        for f,h in plan['config_hashes'].items():assert ops.g.sha(Path(f))==h
        assert ops.inspect('contextra-redis-1')['Id']==plan['redis_id']
        for other in plan['other_streams']:
            now=api('/api/v1/streams/'+other['stream_id']);assert now['state']==other['state'] and now.get('run_id')==other.get('run_id')
        finalsender=sender();assert finalsender['executionId']==send['executionId'] and finalsender['anchor']['frame_time_anchor']==send['anchor']['frame_time_anchor']
        state.update(stage='PASS',before=before,after=after,elapsed_seconds=t1-t0,result_fps=result_fps,redis_observed_fps=redis_fps,redis_entries_delta=x1['entries-added']-x0['entries-added'],redis_frames_checked=len(seen),redis_samples_stored=len(frames),redis_length_range=[min(lengths),max(lengths)],ttl_ms_range=[min(ttls),max(ttls)],events_observed=len(event_ids),red_object_observations=red,publisher=pub,resources=[r0,r1],gpu=gp,metrics=after['metadata_path']['performance'],settings=conf,sender_final=finalsender)
        save('frames.json',list(frames.values()))
        assert (R/'frames.json').stat().st_size<caps['evidence_bytes']
        success=True
    except Exception:
        state.update(stage='FAIL',error=traceback.format_exc())
    finally:
        signal.alarm(0)
        if metrics_on:
            try:
                ledger['metrics_off']=1;save('ledger.json',ledger)
                api(P+'/metrics','POST',dict(run_generation=state['generation'],enabled=False,seconds=1))
            except Exception:state['metrics_disable_error']=traceback.format_exc();success=False
        if not success and started:
            ledger['stop']=1;save('ledger.json',ledger)
            try:
                state['stop_response']=api(P+'/stop','POST');deadline=time.monotonic()+40
                while time.monotonic()<deadline:
                    last=api(P)
                    if last['state']=='stopped' and not last.get('cleanup_pending'):break
                    time.sleep(.3)
                assert last['state']=='stopped' and not last.get('cleanup_error'),'CLEANUP_FAILED'
                state['cleanup']=last
                if ledger['patch']:
                    restored=api(P+'/settings')['configured']
                    state['restore']=api(P+'/settings','PATCH',dict(expected_revision=restored['revision'],frame_time_anchor=plan['configured']['frame_time_anchor']))
            except Exception:state['cleanup_error']=traceback.format_exc()
        state['final']=api(P);state['retained_running']=success and state['final']['state']=='running'
        save('live.json',state);save('ledger.json',ledger)
    print(json.dumps({k:state.get(k) for k in ['stage','result_fps','redis_observed_fps','redis_frames_checked','redis_length_range','ttl_ms_range','events_observed','red_object_observations','retained_running','error','cleanup_error']},ensure_ascii=False))

if __name__=='__main__':{'gate':gate,'run':run}[sys.argv[1]]()