import importlib.util
from pathlib import Path
import sys
import time
import yaml

R=Path(__file__).parent
R.chmod(0o700)
s=importlib.util.spec_from_file_location('previous','/tmp/codex/ltb1021/2026-09-15_contextra-publisher-timeline_r01/ops.py')
previous=importlib.util.module_from_spec(s);s.loader.exec_module(previous)
previous.R=R;previous.old.R=R;previous.old.prev.R=R;p=previous.p;p.R=R
p.FILES=['app/metadata_publisher.py','app/channel_publisher.py','app/metadata_runtime.py','app/det_result_path.py',
         'docker-compose.yml','tests/test_channel_publisher.py']

def apply(label):
    plan=p.read('plan.json');b=p.read(label+'-build.json')
    assert p.read(label+'-image.json')['success'] and not (R/(label+'-apply-start.json')).exists()
    assert p.sha(p.prior.REG)==plan['registry_hash']
    assert p.cmd(['git','rev-parse','HEAD']).strip()==plan['head']
    assert all(p.sha(p.REPO/path)==digest for path,digest in b['hashes'].items())
    assert p.sha(p.ACTIVE)==plan['active_hash'] and p.sha(p.ENV)==plan['env_hash']
    assert all(q['stream_id'] in p.CHANNELS.values() or q['state']=='stopped' for q in p.api('/api/v1/streams')['streams'])
    p.save('runtime-bounds.json',dict(window_seconds=24,first_frame_seconds=90,stop_seconds=90,
        maximum_selected_frames_per_channel=1500,helper_cpu_seconds=60,helper_ram_gib=4,
        analytics_ram_gib=12,gpu_mib=16000,disk_floor_gib=15,log_mib=30,
        mutations='Stop each target once; recreate Analytics once; Start each target once',
        rollback='CAS restore active/env backups and previous image; sender and Redis unchanged; no repeated failure',
        started=time.time(),deadline=time.time()+2100))
    for sid in p.CHANNELS.values():
        if not p.quiescent(p.prior.state(sid)):p.stop(sid,label)
    current=p.base.inspect('contextra-analytics-1');ah,eh=p.sha(p.ACTIVE),p.sha(p.ENV)
    p.save(label+'-apply-start.json',dict(before=current,active_hash=ah,env_hash=eh,
        active_text=p.ACTIVE.read_text(),env_text=p.ENV.read_text(),new_image=b['image'],
        worker_mode='channel',active_worker_limit=16,rollback_image=plan['analytics']['Image']))
    cfg=yaml.safe_load(p.ACTIVE.read_text());cfg['services']['analytics']['image']=b['image']
    changes={'CONTEXTRA_METADATA_WORKER_MODE':'channel','CONTEXTRA_METADATA_ACTIVE_WORKER_LIMIT':'16'}
    cfg['services']['analytics'].setdefault('environment',{}).update(changes)
    p.atomic(p.ACTIVE,yaml.safe_dump(cfg).encode(),ah)
    lines=p.ENV.read_text().splitlines(keepends=True)
    for key,value in dict(changes,CONTEXTRA_IMAGE=b['tag']).items():
        idx=[i for i,v in enumerate(lines) if v.startswith(key+'=')];assert len(idx)<=1
        if idx:lines[idx[0]]=key+'='+value+'\n'
        else:lines.append(key+'='+value+'\n')
    p.atomic(p.ENV,''.join(lines).encode(),eh)
    compose=['docker','compose','--env-file',str(p.ENV),'-f',str(p.REPO/'docker-compose.yml'),'-f',str(p.ACTIVE)]
    p.ledger('apply',label=label,image=b['image'],worker_mode='channel',active_worker_limit=16)
    p.cmd(compose+['config','--quiet']);p.cmd(compose+['up','--no-start','--no-deps','--force-recreate','--pull','never','analytics'])
    p.cmd(['docker','start','contextra-analytics-1'])
    end=time.monotonic()+120
    while time.monotonic()<end:
        cur=p.base.inspect('contextra-analytics-1')
        if cur['State'].get('Health',{}).get('Status')=='healthy':break
        time.sleep(1)
    else:raise RuntimeError('READINESS_FAILED')
    assert cur['Image']==b['image']
    expected=dict(x.split('=',1) for x in plan['analytics']['Config']['Env']);expected.update(changes)
    assert dict(x.split('=',1) for x in cur['Config']['Env'])==expected
    assert p.base.inspect('contextra-redis-1')['Id']==plan['redis']
    assert all(q['state']=='stopped' for q in p.api('/api/v1/streams')['streams'])
    m=p.api('/api/v1/metadata/status');p.save('stopped-publisher.json',m)
    assert m['metrics']['workers']==0 and m['metrics']['workerMode']=='channel'
    for src,sid in p.CHANNELS.items():
        assert p.api('/api/v1/streams/'+sid+'/settings')['configured']==plan['settings'][src]['configured']
        p.ledger('start',stream=sid,label=label);p.api('/api/v1/streams/'+sid+'/start','POST')
        end=time.monotonic()+90
        while time.monotonic()<end:
            q=p.prior.state(sid)
            if q['state']=='running' and q.get('metadata_path',{}).get('processed',0)>0:break
            if q['state']=='error':raise RuntimeError('START_FAILED')
            time.sleep(1)
        else:raise RuntimeError('FIRST_FRAME_TIMEOUT')
    p.save(label+'-applied.json',dict(states={src:p.prior.state(sid) for src,sid in p.CHANNELS.items()},
                                   publisher=p.api('/api/v1/metadata/status')))
    print('3 channels running; channel workers3/control1; cap16')

if __name__=='__main__':
    mode=sys.argv[1]
    label=sys.argv[2] if len(sys.argv)>2 else 'channel3'
    {'gate':p.gate,'test':lambda:p.tests(label+'-source'),
     'build':lambda:previous.old.prev.build(label),
     'apply':lambda:apply(label),
     'observe':lambda:previous.observe('channel')}[mode]()