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

os.umask(0o077)
R = Path(__file__).parent
spec = importlib.util.spec_from_file_location('prior', '/tmp/codex/ltb1021/2026-09-15_contextra-channel-compare_r01/compare.py')
p = importlib.util.module_from_spec(spec)
spec.loader.exec_module(p)
p.R = R
p.SOURCES.append('rtsp-test-source-03')
resource.setrlimit(resource.RLIMIT_CPU, (60, 60))

def gate():
    p.gate()
    plan = p.read('plan.json')
    plan['limits'].update(window_seconds=24, windows=3, snapshots_per_window=26,
                          raw_frames=250, run_seconds=600, helper_memory_bytes=512*1024**2)
    plan['rollback'] = 'Stop03 once for comparison; preserve01/02 run and settings. Restore03 Start once. If sustained degradation recurs, protective Stop03 once. No sender/config/image changes.'
    p.save('plan.json', plan)
    print(json.dumps({s: {'state':x['state'], 'queue':x['metadata_path']['queued'], 'profile':x['inference_profile'], 'scenarios':x['scenarios']} for s,x in plan['states'].items()}))

def window(name, channels, seconds=24):
    assert not (R / (name+'.json')).exists()
    before = {s:p.state(sid) for s,sid in channels.items()}
    samples=[]; enabled=[]; resources=[]
    try:
        for s,sid in channels.items():
            assert not before[s]['metadata_path']['performance']['detail']
            p.mutation(sid,'metrics',dict(run_generation=before[s]['run_generation'],enabled=True,seconds=45))
            enabled.append(s)
        resources.append(p.resource())
        samples.append(p.snapshot(channels))
        deadline=time.monotonic()+seconds
        while time.monotonic()<deadline:
            time.sleep(min(1,max(0,deadline-time.monotonic())))
            samples.append(p.snapshot(channels))
            assert resource.getrusage(resource.RUSAGE_SELF).ru_maxrss < 512*1024
            assert all(samples[-1]['channels'][s]['state']['run_id']==before[s]['run_id'] for s in channels)
        resources.append(p.resource())
        assert resources[-1]['memory']<12*1024**3 and resources[-1]['mem_available']>4*1024**3
    finally:
        p.save(name+'.json',dict(samples=samples,resources=resources))
        for s in enabled:
            st=p.state(channels[s])
            if st['run_id']==before[s]['run_id']:
                p.mutation(channels[s],'metrics',dict(run_generation=st['run_generation'],enabled=False,seconds=1))
    summarize(name)

def summarize(name):
    data=p.read(name+'.json'); out={}
    for s,a in data['samples'][0]['channels'].items():
        b=data['samples'][-1]['channels'][s];dt=b['at']-a['at']
        ma=a['state']['metadata_path'];mb=b['state']['metadata_path']
        pa=ma['performance'];pb=mb['performance']
        out[s]=dict(seconds=dt, fps={k:round((v-pa['counts'].get(k,0))/dt,3) for k,v in pb['counts'].items()},
                    redis_fps=round((b['redis']['entries']-a['redis']['entries'])/(b['redis']['at']-a['redis']['at']),3),
                    queue=[x['channels'][s]['state']['metadata_path']['queued'] for x in data['samples']],
                    counter_delta={k:mb.get(k,0)-ma.get(k,0) for k in set(ma)|set(mb) if isinstance(ma.get(k),int) and not isinstance(ma.get(k),bool) and isinstance(mb.get(k),int)},
                    durations=pb['durations'],profiles=pb.get('tracking_profiles'),
                    shapes=list({(x['channels'][s]['redis']['frame'].get('frameWidth'),x['channels'][s]['redis']['frame'].get('frameHeight')) for x in data['samples']}),
                    ttl=[min(x['channels'][s]['redis']['pttl'] for x in data['samples']),max(x['channels'][s]['redis']['pttl'] for x in data['samples'])])
    p.save(name+'-summary.json',out)
    print(json.dumps({s:dict(fps=v['fps'],redis_fps=v['redis_fps'],queue=[min(v['queue']),max(v['queue'])],delta=v['counter_delta'],tracking=v['durations'].get('tracking'),cpu=v['durations'].get('tracking_cpu'),profiles=[{k:q[k] for k in ('kernel_wait_ms','wall_ms','thread_cpu_ms')} for q in v['profiles']]) for s,v in out.items()},ensure_ascii=False))

def stop(label):
    plan=p.read('plan.json');sid=plan['channels'][p.SOURCES[2]]
    assert not (R/(label+'-requested.json')).exists()
    st=p.state(sid);assert st['state']=='running'
    p.save(label+'-requested.json',dict(at=time.time(),state=st))
    p.mutation(sid,'stop')
    history=[];deadline=time.monotonic()+90
    while time.monotonic()<deadline:
        q=p.state(sid);history.append(q)
        cleanup=q.get('last_cleanup') or {}
        clean=not q.get('cleanup_pending') and not q.get('cleanup_error') and q.get('desired_state')=='stopped' and (q['state']=='stopped' or (cleanup.get('completed') and cleanup.get('run_generation')==st['run_generation']))
        if clean:break
        time.sleep(1)
    p.save(label+'-result.json',dict(clean=clean,history=history))
    print(json.dumps(dict(clean=clean,state=q['state'],cleanup=q.get('last_cleanup'))))
    assert clean,'No retry. Preserve other channels.'

def start():
    plan=p.read('plan.json');sid=plan['channels'][p.SOURCES[2]]
    assert not (R/'restore-requested.json').exists()
    assert p.state(sid)['state']=='stopped'
    assert p.api('/api/v1/streams/'+sid+'/settings')['configured']==plan['settings'][p.SOURCES[2]]['configured']
    p.save('restore-requested.json',dict(at=time.time()))
    p.mutation(sid,'start');history=[];deadline=time.monotonic()+90;ready=False
    while time.monotonic()<deadline:
        q=p.state(sid);history.append(q)
        if q['state']=='running' and q.get('metadata_path',{}).get('processed',0)>0:ready=True;break
        if q['state']=='error':break
        time.sleep(1)
    p.save('restore-result.json',dict(ready=ready,history=history))
    print('restore_ready='+str(ready));assert ready

if __name__=='__main__':
    mode=sys.argv[1]
    if mode=='gate':gate()
    elif mode=='three':window('three',p.read('plan.json')['channels'])
    elif mode=='two':
        channels={s:sid for s,sid in p.read('plan.json')['channels'].items() if s!=p.SOURCES[2]}
        time.sleep(5)
        assert all(p.state(sid)['run_id']==p.read('plan.json')['states'][s]['run_id'] for s,sid in channels.items())
        window('two',channels)
    elif mode=='restore':start()
    elif mode=='restored':
        time.sleep(5);window('restored',p.read('plan.json')['channels'],12)
    elif mode in ('comparison-stop','protective-stop'):stop(mode)
    elif mode=='final':p.final()