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

os.umask(0o077)
R=Path(__file__).parent;R.chmod(0o700)
REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
sys.path.insert(0,str(REPO));sys.path.insert(0,str(REPO/'tools'))
from test_run_manager import TestRunManager, load_preset
s=importlib.util.spec_from_file_location('prior','/tmp/codex/ltb1021/2026-09-15_contextra-channel-publisher_r01/ops.py')
prior=importlib.util.module_from_spec(s);s.loader.exec_module(prior);p=prior.p
REG=p.prior.REG
ORIGINAL=['rtsp-test-01','web-cam08-test','web-cam01-test']

def save(name,value):
    (R/name).write_text(json.dumps(value,ensure_ascii=False,indent=2));(R/name).chmod(0o600)
def read(name):return json.loads((R/name).read_text())
def ledger(action,**kw):
    with (R/'calls.jsonl').open('a') as f:f.write(json.dumps(dict(at=time.time(),action=action,**kw))+'\n')
def api(path,method='GET',body=None):
    if method!='GET':ledger('request',method=method,path=path)
    return p.api(path,method,body)
def states():return {x['stream_id']:x for x in api('/api/v1/streams')['streams']}
def resources():
    a=p.base.inspect('contextra-analytics-1');pid=a['State']['Pid']
    cg=Path('/sys/fs/cgroup'+Path(f'/proc/{pid}/cgroup').read_text().strip().split('::')[1])
    mem={x.split(':')[0]:int(x.split()[1])*1024 for x in Path('/proc/meminfo').read_text().splitlines()}
    cpu={k:int(v) for k,v in (x.split() for x in (cg/'cpu.stat').read_text().splitlines())}
    host=[int(x) for x in Path('/proc/stat').read_text().splitlines()[0].split()[1:]]
    return dict(at=time.monotonic(),memory=int((cg/'memory.current').read_text()),available=mem['MemAvailable'],
                cpu=cpu,host_cpu=host,gpu=p.prior.ops.prior.prev.gpu(),disk=shutil.disk_usage('/var/lib/docker').free,
                memory_events=(cg/'memory.events').read_text())
def guard():
    r=resources()
    assert r['memory']<10*1024**3 and r['available']>8*1024**3,'RAM_BOUND'
    assert r['disk']>15*1024**3,'DISK_BOUND'
    assert all(x[0]<min(40000,x[1]*.9) for x in r['gpu']),'GPU_MEMORY_BOUND'
    return r

def gate():
    assert not (R/'plan.json').exists()
    st=states();registry=yaml.safe_load(REG.read_text());rows={x['stream_id']:x for x in registry['added']}
    assert all(st[s]['state']=='running' for s in ORIGINAL)
    other=[sid for sid in st if sid not in ORIGINAL];assert len(other)==1,'FOURTH_INPUT_AMBIGUOUS'
    fourth=other[0];assert st[fourth]['state']=='stopped' and rows[fourth]['source_type']=='rtsp'
    settings={sid:api('/api/v1/streams/'+sid+'/settings')['configured'] for sid in st}
    for sid in ORIGINAL:
        c=settings[sid];assert c['tracking']['enabled'] is False and c['intrusion']['enabled'] is False
        assert c['analysisFpsTarget']==10 and not st[sid]['metadata_path']['performance']['detail']
        assert st[sid]['metadata_path']['visual'] is None and not rows[sid]['store'] and not rows[sid]['active_learning']
    inputs=[rows[sid] for sid in ORIGINAL+[fourth]]
    assert len({x['uri'] for x in inputs})==4
    channels=[];temporary=[]
    (R/'presets').mkdir(mode=0o700)
    for group,source in enumerate(inputs,1):
        cfg=settings[source['stream_id']] if group<=3 else settings[ORIGINAL[0]]
        for copy in range(1,5):
            original=group<=3 and copy==1
            sid=source['stream_id'] if original else f'tmp16-0915-g{group}-c{copy}'
            src=source['source_id'] if original else f'tmp16-0915-source-g{group}-c{copy}'
            if not original:
                assert sid not in st and all(x.get('source_id')!=src for x in rows.values())
            row=dict(group=group,copy=copy,stream_id=sid,source_id=src,original=original,
                     uri_hash=hashlib.sha256(source['uri'].encode()).hexdigest(),origin=source['stream_id'])
            channels.append(row)
            if original:continue
            opts={k:cfg[k] for k in ('display_score','display_categories','debug_all_candidates','intrusion')}
            opts.update(tracking={'enabled':False},analysisFpsTarget=10,
                        frame_time_anchor=source.get('frame_time_anchor'))
            opts['intrusion']=dict(opts['intrusion'],enabled=False)
            preset=dict(version=1,camera=dict(stream_id=sid,source_id=src,server_id=source['server_id'],uri=source['uri']),
                        settings=opts,debug_metrics=False,wait_seconds=90)
            file=R/'presets'/(sid+'.yaml');file.write_text(yaml.safe_dump(preset,allow_unicode=True));file.chmod(0o600)
            load_preset(file);temporary.append(sid)
    assert len(channels)==16 and len(temporary)==13
    baseline=guard();a=p.base.inspect('contextra-analytics-1')
    save('plan.json',dict(at=time.time(),head=subprocess.check_output(['git','rev-parse','HEAD'],cwd=REPO,text=True).strip(),
        registry=registry,registry_hash=p.sha(REG),states=st,settings=settings,channels=channels,temporary=temporary,
        analytics=a,inventory=p.base.inventory(),sender=p.prior.ops.prior.prev.sender(),resources=baseline,
        config_hashes={str(x):p.sha(x) for x in [p.ACTIVE,p.ENV]},
        caps=dict(total_seconds=1200,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 each added channel once then DELETE only after cleanup; preserve originals and registered fourth. Publisher stopped heartbeat registry remains by existing API contract. No container restart.'))
    print('Gate: original3 running, registered fourth stopped; 13 temporary presets; 4 inputs x4. No mutation.',flush=True)

class LoggedApi:
    def call(self,method,path,body=None):return api(path,method,body)

def redis_sample(row,frame=False):
    key='contextra:metadata:v1:camera:'+hashlib.sha256(row['source_id'].encode()).hexdigest()+':frames'
    q=p.base.m.query;t=time.monotonic();info=q('reader','XINFO','STREAM',key)
    if not isinstance(info,dict):return dict(at=t,missing=True,error=str(info)[:100])
    v=dict(at=t,entries=info['entries-added'],length=info['length'],pttl=q('reader','PTTL',key))
    if frame:
        data=q('reader','XREVRANGE',key,'+','-','COUNT','1')
        fields=data[0][1]
        if not isinstance(fields,dict):fields=dict(zip(fields[::2],fields[1::2]))
        v['frame']=json.loads(fields['json'])
    return v

def snapshot(plan,frames=False):
    st=states();at=time.monotonic();result={}
    for row in plan['channels']:
        sid=row['stream_id'];v=st[sid];path=v.get('metadata_path') or {};perf=path.get('performance') or {}
        result[sid]=dict(state=v['state'],run_id=v.get('run_id'),generation=v.get('run_generation'),
            error=v.get('last_error'),cleanup_error=v.get('cleanup_error'),
            processed=path.get('processed',0),queued=path.get('queued',0),stale=path.get('stale',0),
            drop=path.get('queue_drop',0),rejected=path.get('publisher_rejected',0),
            counts=perf.get('counts',{}),observation_missed=perf.get('observation_missed',0),
            stage_counts=v.get('roles',{}).get('det',{}).get('stage_counts',{}),
            redis=redis_sample(row,frames))
    return dict(at=at,channels=result,publisher=api('/api/v1/metadata/status')['metrics'],resources=guard())

def cleanup(plan):
    reports={}
    for sid in reversed(plan['temporary']):
        try:
            st=states()
            if sid not in st:continue
            q=st[sid]
            if q['state']!='stopped' or q.get('cleanup_pending'):
                if sid in STOP_SENT:raise RuntimeError('STOP_ALREADY_SENT')
                STOP_SENT.add(sid);api('/api/v1/streams/'+sid+'/stop','POST')
                end=time.monotonic()+90
                while time.monotonic()<end:
                    q=api('/api/v1/streams/'+sid)
                    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')
            reports[sid]=dict(cleanup=q.get('last_cleanup'),removed=api('/api/v1/streams/'+sid,'DELETE'))
            print('Removed',sid,flush=True)
        except Exception as e:
            reports[sid]=dict(error=type(e).__name__,detail=str(e)[:200])
            save('cleanup-partial.json',reports)
    save('cleanup.json',reports)

STOP_SENT=set()
def execute():
    plan=read('plan.json');assert not (R/'execution-start.json').exists()
    assert p.sha(REG)==plan['registry_hash']
    save('execution-start.json',dict(at=time.time()));start=time.monotonic();samples=[];result='NOT_MEASURED'
    profile=False
    try:
        for sid in plan['temporary']:
            guard();assert time.monotonic()-start<900,'START_BUDGET'
            preset=load_preset(R/'presets'/(sid+'.yaml'))
            last_check=[0.]
            def bounded_sleep(seconds):
                if time.monotonic()-last_check[0]>3:
                    guard();last_check[0]=time.monotonic()
                time.sleep(seconds)
            manager=TestRunManager(preset,api=LoggedApi(),sleep=bounded_sleep)
            value=manager.execute('start');save(sid+'-started.json',value)
            print('Running',sid,flush=True)
        st=states();assert sum(x['state']=='running' for x in st.values())==16
        assert api('/api/v1/metadata/status')['metrics']['workers']==16
        save('sixteen-started.json',dict(at=time.time(),states=st))
        for _ in range(5):time.sleep(1);guard()
        # Counters are observed for60s; bounded detail expires independently.
        api('/api/v1/metadata/metrics','POST',dict(enabled=True,seconds=60));profile=True
        samples.append(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=snapshot(plan);samples.append(sample)
                assert all(x['state']=='running' and not x['error'] for x in sample['channels'].values()),'CHANNEL_FAILED'
                assert all(x['run_id']==samples[0]['channels'][sid]['run_id'] for sid,x in sample['channels'].items()),'RUN_CHANGED'
                if len(samples)>=4 and all(all(x['queued']>=24 for x in s['channels'].values()) for s in samples[-3:]):
                    raise RuntimeError('SUSTAINED_QUEUE_OVERLOAD')
        samples.append(snapshot(plan,True));result='MEASURED'
    except Exception as e:
        save('failure.json',dict(at=time.time(),error=type(e).__name__,detail=str(e)[:1000],trace=traceback.format_exc()))
        print('Stopped trial:',type(e).__name__,str(e)[:150],flush=True)
    finally:
        save('window.json',dict(result=result,samples=samples))
        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__,detail=str(e)[:200]))
        cleanup(plan)
        final=states();reg=yaml.safe_load(REG.read_text())
        save('final.json',dict(states=final,registry=reg,publisher=api('/api/v1/metadata/status'),resources=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']}))
        print('Final states:',[(sid,v['state']) for sid,v in final.items()],flush=True)

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