import importlib.util
import json
import os
from pathlib import Path
import shutil
import sys
import time
import yaml

os.umask(0o077)
R=Path(__file__).parent
s=importlib.util.spec_from_file_location('prev','/tmp/codex/ltb1021/2026-09-15_contextra-active-tracking_r01/ops.py')
prev=importlib.util.module_from_spec(s);s.loader.exec_module(prev)
prev.R=R;p=prev.previous;p.R=R
p.CHANNELS['rtsp-test-source-03']='web-cam01-test'
p.FILES=['app/byte_tracking.py','app/stream_settings.py','app/api/streams.py','app/stream_manager.py','app/det_result_path.py','tools/test_run_manager.py','tests/test_tracking_disabled.py','tests/test_test_run.py']
TARGET='web-cam08-test'

def gate():
    assert not (R/'plan.json').exists()
    states={src:p.prior.state(sid) for src,sid in p.CHANNELS.items()}
    assert all(q['state'] in ('running','stopped') for q in states.values())
    for n,f in [('active',p.ACTIVE),('env',p.ENV),('registry',p.prior.REG)]:shutil.copy2(f,R/(n+'.before'))
    preset=Path('/workspace/contextra-runtime-config/test-run/presets/web-cam08.yaml')
    shutil.copy2(preset,R/'preset.before')
    p.save('plan.json',dict(started=time.time(),head=p.cmd(['git','rev-parse','HEAD']).strip(),
        analytics=p.base.inspect('contextra-analytics-1'),inventory=p.base.inventory(),redis=p.base.inspect('contextra-redis-1')['Id'],
        sender=p.prior.ops.prior.prev.sender(),states=states,settings={s:p.api('/api/v1/streams/'+sid+'/settings') for s,sid in p.CHANNELS.items()},
        registry_hash=p.sha(p.prior.REG),active_hash=p.sha(p.ACTIVE),env_hash=p.sha(p.ENV),preset=str(preset),preset_hash=p.sha(preset),resources=p.prior.resource(),
        limits=dict(preparation_seconds=1200,runtime_seconds=2100,build_seconds=1200,disk_gib=15,window_seconds=24,helper_ram_gib=4,analytics_ram_gib=12,gpu_mib=16000,source_timeout_seconds=90),
        scope='Latest user: ONLY02 running. Stop01 and keep03 stopped; no3-channel experiment. Preserve01/03 config.',
        rollback='Original image/env/active and registry/preset snapshots retained. No repeated failed Stop. CAS restore only necessary changes, keep non-target services/sender/Redis.'))
    print('gate', {k:v['state'] for k,v in states.items()})

def apply():
    plan=p.read('plan.json');b=p.read('off-build.json')
    assert p.read('off-image.json')['success'] and not (R/'off-apply-start.json').exists()
    assert p.sha(p.prior.REG)==plan['registry_hash']
    assert all(q['stream_id'] in p.CHANNELS.values() or q['state']=='stopped' for q in p.api('/api/v1/streams')['streams'])
    for sid in p.CHANNELS.values():
        if not p.quiescent(p.prior.state(sid)):p.stop(sid,'off')
    assert all(p.quiescent(p.prior.state(sid)) for sid in p.CHANNELS.values())
    assert p.sha(p.ACTIVE)==plan['active_hash'] and p.sha(p.ENV)==plan['env_hash']
    p.save('off-apply-start.json',dict(image=b,restore_image=plan['analytics']['Image'],at=time.time()))
    cfg=yaml.safe_load(p.ACTIVE.read_text());cfg['services']['analytics']['image']=b['image']
    p.atomic(p.ACTIVE,yaml.safe_dump(cfg).encode(),plan['active_hash'])
    lines=p.ENV.read_text().splitlines(keepends=True);idx=[i for i,v in enumerate(lines) if v.startswith('CONTEXTRA_IMAGE=')];assert len(idx)==1
    lines[idx[0]]='CONTEXTRA_IMAGE='+b['tag']+'\n';p.atomic(p.ENV,''.join(lines).encode(),plan['env_hash'])
    compose=['docker','compose','--env-file',str(p.ENV),'-f',str(p.REPO/'docker-compose.yml'),'-f',str(p.ACTIVE)]
    p.ledger('image_apply',image=b['image'])
    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'])
    deadline=time.monotonic()+120
    while time.monotonic()<deadline:
        cur=p.base.inspect('contextra-analytics-1')
        if cur['State'].get('Health',{}).get('Status')=='healthy':break
        time.sleep(1)
    else:raise RuntimeError('readiness')
    assert cur['Image']==b['image'] and all(q['state']=='stopped' for q in p.api('/api/v1/streams')['streams'])
    assert p.base.inspect('contextra-redis-1')['Id']==plan['redis']
    p.save('off-ready.json',dict(at=time.time(),image=cur['Image']))
    preset=Path(plan['preset']);assert p.sha(preset)==plan['preset_hash']
    data=yaml.safe_load(preset.read_text());assert data['camera']['stream_id']==TARGET
    data.setdefault('settings',{})['tracking']={'enabled':False}
    data['settings'].setdefault('intrusion',{})['enabled']=False;data['debug_metrics']=False
    p.atomic(preset,yaml.safe_dump(data,allow_unicode=True,sort_keys=False).encode(),plan['preset_hash'])
    p.ledger('cli_start02',preset_hash=p.sha(preset))
    p.cmd(['python3',str(p.REPO/'tools/test_run.py'),'start',str(preset)],timeout=110)
    states={src:p.prior.state(sid) for src,sid in p.CHANNELS.items()}
    assert states['rtsp-test-source-02']['state']=='running'
    assert all(q['state']=='stopped' for src,q in states.items() if src!='rtsp-test-source-02')
    p.save('off-applied.json',states);print('ONLY02 running,01/03 stopped')

if __name__=='__main__':
    {'gate':gate,'test':lambda:p.tests('off-source'),'build':lambda:prev.build('off'),'apply':apply}[sys.argv[1]]()