import importlib.util
import json
import os
from pathlib import Path
import resource
import subprocess
import sys
import time
import traceback
import yaml

os.umask(0o077)
resource.setrlimit(resource.RLIMIT_CPU,(120,120))
R=Path(__file__).parent;R.chmod(0o700)
s=importlib.util.spec_from_file_location('old','/tmp/codex/ltb1021/2026-09-15_contextra-sixteen-off_r01/run.py')
old=importlib.util.module_from_spec(s);s.loader.exec_module(old);old.R=R
p=old.p;REPO=old.REPO;REG=old.REG;ORIGINAL=old.ORIGINAL
save=old.save;read=old.read;ledger=old.ledger
from test_run_manager import load_preset, select_changes
PHASE='gate';RESOURCE_SAMPLES=[]

def api(path,method='GET',body=None):
    started=time.monotonic()
    if method!='GET':ledger('request',method=method,path=path,phase=PHASE)
    try:return p.api(path,method,body)
    except Exception as e:
        ledger('api_error',method=method,path=path,phase=PHASE,elapsed=time.monotonic()-started,error=type(e).__name__)
        raise
    finally:
        if method!='GET':ledger('request_finished',method=method,path=path,phase=PHASE,elapsed=time.monotonic()-started)
old.api=api
def states():return {x['stream_id']:x for x in api('/api/v1/streams')['streams']}
old.states=states
def guard():
    r=old.guard();RESOURCE_SAMPLES.append(dict(r,phase=PHASE));return r

def gate():
    assert not (R/'plan.json').exists()
    st=states();reg=yaml.safe_load(REG.read_text());regs={x['stream_id']:x for x in reg['added']}
    assert all(st[sid]['state']=='running' for sid in ORIGINAL)
    assert all(v['state']=='stopped' for sid,v in st.items() if sid not in ORIGINAL)
    settings={sid:api('/api/v1/streams/'+sid+'/settings')['configured'] for sid in st}
    pub=api('/api/v1/metadata/status')['metrics']
    assert pub['workerMode']=='channel' and pub['workers']==3 and pub['workerLimit']==16
    assert pub['channels']+13<=64 and not pub['profile']['enabled']
    source_ids={x['source_id'] for x in regs.values()}
    channels=[];temporary=[];(R/'presets').mkdir(mode=0o700)
    for group,(sid,count) in enumerate(zip(ORIGINAL,(6,5,5)),1):
        src=regs[sid];cfg=settings[sid]
        assert not cfg['tracking']['enabled'] and not cfg['intrusion']['enabled'] and cfg['analysisFpsTarget']==10
        assert not src['store'] and not src['active_learning'] and st[sid]['metadata_path']['visual'] is None
        for copy in range(1,count+1):
            orig=copy==1;target=sid if orig else f'tmp16b-2128-g{group}-c{copy}'
            source=src['source_id'] if orig else f'tmp16b-2128-source-g{group}-c{copy}'
            if not orig:assert target not in st and source not in source_ids
            row=dict(group=group,copy=copy,stream_id=target,source_id=source,original=orig,origin=sid)
            channels.append(row)
            if orig:continue
            values={k:cfg[k] for k in ('display_score','display_categories','debug_all_candidates','intrusion')}
            values.update(tracking={'enabled':False},analysisFpsTarget=10,frame_time_anchor=src.get('frame_time_anchor'))
            preset=dict(version=1,camera=dict(stream_id=target,source_id=source,server_id=src['server_id'],uri=src['uri']),
                        settings=values,debug_metrics=False,wait_seconds=90)
            file=R/'presets'/(target+'.yaml');file.write_text(yaml.safe_dump(preset,allow_unicode=True));file.chmod(0o600)
            load_preset(file);temporary.append(target)
    assert len(channels)==16 and len(temporary)==13
    save('plan.json',dict(at=time.time(),head=subprocess.check_output(['git','rev-parse','HEAD'],cwd=REPO,text=True).strip(),
        registry=reg,registry_hash=p.sha(REG),states=st,settings=settings,channels=channels,temporary=temporary,
        analytics=p.base.inspect('contextra-analytics-1'),inventory=p.base.inventory(),sender=p.prior.ops.prior.prev.sender(),
        resources=guard(),publisher=pub,config_hashes={str(x):p.sha(x) for x in [p.ACTIVE,p.ENV]},
        caps=dict(total_seconds=1200,start_phase_seconds=900,http_timeout_seconds=5,window_seconds=60,
                  first_frame_seconds=90,helper_cpu_seconds=120,ram_gib=10,min_host_available_gib=8,
                  gpu_mib=40000,disk_floor_gib=15,raw_evidence_mib=30),
        restore='Stop temporary once, basic list cleanup confirmation, then DELETE. Preserve original run/settings/sender/Redis. Stopped publisher records retained.'))
    print('Gate PASS: existing3,13 new identities,6/5/5 inputs; publisher registry',pub['channels'],'->',pub['channels']+13,flush=True)

def prepare(plan):
    global PHASE
    PHASE='prepare'
    for sid in plan['temporary']:
        guard();preset=load_preset(R/'presets'/(sid+'.yaml'))
        body=dict(preset['camera'],source_type='rtsp',site_id=None,store=False,active_learning=False,
                  autostart=False,scenarios='off',analysisFpsTarget=10,inference_profile={'roles':['det']})
        registered=api('/api/v1/streams?start=false','POST',body)
        assert registered['state']=='stopped'
        current=api('/api/v1/streams/'+sid+'/settings')['configured']
        changes=select_changes(preset['settings'],current)
        if changes:api('/api/v1/streams/'+sid+'/settings','PATCH',dict(expected_revision=current['revision'],**changes))
        current=api('/api/v1/streams/'+sid+'/settings')['configured']
        assert not select_changes(preset['settings'],current)
        save(sid+'-prepared.json',dict(configured=current))
    st=states();assert all(st[sid]['state']=='stopped' for sid in plan['temporary'])
    assert all(st[sid]['run_id']==plan['states'][sid]['run_id'] for sid in ORIGINAL)
    save('all-prepared.json',dict(at=time.time(),states=st))
    print('All13 registered/configured; temporary Start0. Starting now.',flush=True)

def compact():
    st=states();return dict(at=time.monotonic(),channels={sid:dict(state=q['state'],run_id=q.get('run_id'),generation=q.get('run_generation'),
        processed=q.get('metadata_path',{}).get('processed',0),queued=q.get('metadata_path',{}).get('queued'),
        counts=q.get('metadata_path',{}).get('performance',{}).get('counts',{}),error=q.get('last_error')) for sid,q in st.items()})

def cleanup(plan):
    global PHASE
    PHASE='cleanup';reports={}
    for sid in reversed(plan['temporary']):
        try:
            st=states()
            if sid not in st:continue
            q=st[sid];identity=(q.get('run_id'),q.get('run_generation'));report={}
            if q['state']!='stopped' or q.get('cleanup_pending'):
                try:report['stop_response']=api('/api/v1/streams/'+sid+'/stop','POST')
                except Exception as e:report['stop_response_error']=type(e).__name__
                end=time.monotonic()+90
                while time.monotonic()<end:
                    q=states()[sid]
                    assert (q.get('run_id'),q.get('run_generation'))==identity,'RUN_CHANGED'
                    if q['state']=='stopped' and not q.get('cleanup_pending') and not q.get('cleanup_error'):break
                    if q.get('cleanup_error'):raise RuntimeError('CLEANUP_ERROR')
                    time.sleep(.5)
                else:raise RuntimeError('CLEANUP_TIMEOUT')
            assert q['state']=='stopped' and not q.get('cleanup_pending') and not q.get('cleanup_error')
            if q.get('run_id'):
                assert q['last_cleanup']['completed'] and q['last_cleanup']['generation']==q['run_generation']
            report['cleanup']=q.get('last_cleanup');report['removed']=api('/api/v1/streams/'+sid,'DELETE');reports[sid]=report
        except Exception as e:
            reports[sid]=dict(error=type(e).__name__,detail=str(e)[:300])
        save('cleanup-progress.json',reports)
    save('cleanup.json',reports)

def execute():
    global PHASE
    plan=read('plan.json');assert not (R/'execution-start.json').exists() and p.sha(REG)==plan['registry_hash']
    assert all(p.sha(Path(x))==h for x,h in plan['config_hashes'].items())
    save('execution-start.json',dict(at=time.time()));startup=[];samples=[];profile=False;result='NOT_MEASURED';started=time.monotonic()
    try:
        prepare(plan);PHASE='start';startup.append(compact())
        for sid in plan['temporary']:
            guard();assert time.monotonic()-started<900,'START_BUDGET'
            api('/api/v1/streams/'+sid+'/start','POST');end=time.monotonic()+90
            last_guard=0
            while time.monotonic()<end:
                current=states();q=current[sid]
                if q['state']=='running' and q.get('metadata_path',{}).get('processed',0)>0:break
                if q.get('last_error') or q['state']=='error':raise RuntimeError('START_FAILED')
                if time.monotonic()-last_guard>3:guard();last_guard=time.monotonic()
                time.sleep(.5)
            else:raise RuntimeError('FIRST_FRAME_TIMEOUT')
            save(sid+'-started.json',q);startup.append(compact());save('startup-samples.json',startup)
            print('Running',sum(x['state']=='running' for x in current.values()),'of16',flush=True)
        st=states();pub=api('/api/v1/metadata/status')['metrics']
        assert sum(x['state']=='running' for x in st.values())==16 and pub['workers']==16
        save('sixteen-started.json',dict(states=st,publisher=pub));PHASE='warmup'
        for _ in range(5):guard();time.sleep(1)
        PHASE='measure';profile=True;api('/api/v1/metadata/metrics','POST',dict(enabled=True,seconds=60))
        samples.append(old.snapshot(plan,True));deadline=samples[0]['at']+60
        while time.monotonic()<deadline:
            time.sleep(min(5,max(0,deadline-time.monotonic())))
            if time.monotonic()<deadline-.5:
                sample=old.snapshot(plan);samples.append(sample)
                assert all(q['state']=='running' and not q['error'] for q in sample['channels'].values()),'CHANNEL_FAILED'
                assert all(q['run_id']==samples[0]['channels'][sid]['run_id'] for sid,q in sample['channels'].items()),'RUN_CHANGED'
                if len(samples)>=4 and all(all(q['queued']>=24 for q in x['channels'].values()) for x in samples[-3:]):
                    raise RuntimeError('SUSTAINED_QUEUE_OVERLOAD')
        samples.append(old.snapshot(plan,True));result='MEASURED'
    except Exception as e:
        save('failure.json',dict(at=time.time(),phase=PHASE,error=type(e).__name__,detail=str(e)[:1000],trace=traceback.format_exc()))
        try:save('failure-state.json',compact())
        except Exception as extra:save('failure-state-error.json',dict(error=type(extra).__name__))
        print('Trial stopped:',PHASE,type(e).__name__,str(e)[:150],flush=True)
    finally:
        save('window.json',dict(result=result,samples=samples));save('startup-samples.json',startup)
        if profile:
            try:
                save('publisher-profile.json',api('/api/v1/metadata/status'))
                api('/api/v1/metadata/metrics','POST',dict(enabled=False,seconds=1))
            except Exception as e:save('profile-finish-error.json',dict(error=type(e).__name__))
        cleanup(plan)
        save('resource-samples.json',RESOURCE_SAMPLES)
        final=dict(states=states(),registry=yaml.safe_load(REG.read_text()),publisher=api('/api/v1/metadata/status'),
            resources=old.resources(),inventory=p.base.inventory(),sender=p.prior.ops.prior.prev.sender(),
            original_settings={sid:api('/api/v1/streams/'+sid+'/settings')['configured'] for sid in plan['states']})
        save('final.json',final)
        print('Final:',[(k,v['state']) for k,v in final['states'].items()],'workers',final['publisher']['metrics']['workers'],flush=True)

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