import importlib.util
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tarfile
import tempfile
import time
import yaml

os.umask(0o077)
R=Path(__file__).parent;R.chmod(0o700)
s=importlib.util.spec_from_file_location('prior','/tmp/codex/ltb1021/2026-09-15_contextra-channel-compare_r01/compare.py')
prior=importlib.util.module_from_spec(s);s.loader.exec_module(prior)
base=prior.ops.base;api=prior.api;REPO=prior.ops.REPO
ACTIVE=Path('/workspace/contextra-runtime-config/metadata-active/compose.yaml')
ENV=Path('/workspace/contextra-runtime-config/runtime.env')
CHANNELS={'rtsp-test-source-01':'rtsp-test-01','rtsp-test-source-02':'web-cam08-test'}
FILES=['app/analysis_metrics.py','app/byte_tracking.py','app/stream_manager.py','app/tracking_probe.py','tests/test_tracking_probe.py']
def save(n,v):(R/n).write_text(json.dumps(v,indent=2,ensure_ascii=False))
def read(n):return json.loads((R/n).read_text())
def sha(p):return base.g.sha(p)
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 cmd(args,timeout=120):
    p=subprocess.run(args,capture_output=True,text=True,timeout=timeout)
    with (R/'commands.log').open('a') as f:f.write(p.stdout[-16000:]+p.stderr[-16000:])
    if p.returncode:raise RuntimeError('COMMAND_FAILED: '+p.stderr[-500:])
    return p.stdout
def atomic(path,data,expected):
    assert sha(path)==expected
    stat=path.stat();fd,tmp=tempfile.mkstemp(prefix='.tracking-',dir=path.parent)
    try:
        os.fchmod(fd,stat.st_mode&0o777);os.fchown(fd,stat.st_uid,stat.st_gid)
        with os.fdopen(fd,'wb') as f:f.write(data);f.flush();os.fsync(f.fileno())
        assert sha(path)==expected;os.replace(tmp,path)
    finally:
        if os.path.exists(tmp):os.unlink(tmp)
def gate():
    assert not (R/'plan.json').exists()
    reg=yaml.safe_load(prior.REG.read_text())
    for src,sid in CHANNELS.items():assert any(x.get('source_id')==src and x['stream_id']==sid for x in reg['added'])
    states={src:prior.state(sid) for src,sid in CHANNELS.items()}
    assert all(x['state']=='running' for x in states.values())
    for n,p in [('active',ACTIVE),('env',ENV),('registry',prior.REG)]:shutil.copy2(p,R/(n+'.before'))
    save('plan.json',dict(started=time.time(),head=cmd(['git','rev-parse','HEAD']).strip(),
        analytics=base.inspect('contextra-analytics-1'),inventory=base.inventory(),
        redis=base.inspect('contextra-redis-1')['Id'],sender=prior.ops.prior.prev.sender(),
        settings={src:api('/api/v1/streams/'+sid+'/settings') for src,sid in CHANNELS.items()},states=states,
        registry_hash=sha(prior.REG),active_hash=sha(ACTIVE),env_hash=sha(ENV),resources=prior.resource(),
        caps=dict(preparation_seconds=1200,runtime_seconds=2100,build_seconds=1200,free_disk_gib=15,
            sample_seconds=18,profile_frames_per_channel=3,first_frame_seconds=90,stop_seconds=90,
            helper_ram_gib=4,analytics_ram_gib=12,gpu_mib=16000,raw_log_mib=30),
        restore='Both channel settings unchanged. Restore original image/env/compose with CAS if correction regresses. No sender/Redis action. No repeated failed Stop; preserve first error and assess other channels before safe container stop.'))
    print('Gate PASS; plan and original compose/env/registry preserved',flush=True)
def tests(label,image=None):
    image=image or read('plan.json')['analytics']['Image'];root='/opt/contextra' if image!=read('plan.json')['analytics']['Image'] else '/source'
    old=read_old_cmd();mount=old[old.index('--mount')+1]
    env=dict(x.split('=',1) for x in read('plan.json')['analytics']['Config']['Env'])
    args=['docker','run','--rm','--pull','never','--runtime','runc','--network','none','--name','codex-tracking-'+label,
        '--cpus','2','--memory','4g','--memory-swap','4g','-w',root,'-e','OPENBLAS_NUM_THREADS=1',
        '-e','PYTHONDONTWRITEBYTECODE=1','-e','PYTHONPATH='+root+':/tests:/opt/contextra',
        '-e','LD_LIBRARY_PATH=/driver:'+env.get('LD_LIBRARY_PATH',''),'-e','GST_PLUGIN_PATH='+env.get('GST_PLUGIN_PATH',''),
        '-e','DATABASE_URL=postgresql://fixture:fixture@127.0.0.1:1/fixture','-e','REDIS_STREAMS_ENABLED=false',
        '--mount',mount,'-v',str(REPO)+':/source:ro','-v',str(REPO/'tests')+':/tests:ro',
        '-v',str(R)+':/evidence','--entrypoint','timeout',image,'90','python3','/evidence/verify.py',label]
    ledger('test',label=label,command=args)
    p=subprocess.run(args,capture_output=True,text=True,timeout=110);(R/(label+'.log')).write_text(p.stdout+p.stderr)
    print(label,p.returncode,p.stdout[-1200:],flush=True);assert p.returncode==0
def read_old_cmd():
    return json.loads(Path('/tmp/codex/ltb1021/2026-09-14_contextra-intrusion-preparation_r01/source-test-started.json').read_text())['command']
def build(label):
    assert not (R/(label+'-build.json')).exists()
    hashes={p:sha(REPO/p) for p in FILES};import hashlib
    digest=hashlib.sha256(json.dumps(hashes,sort_keys=True).encode()).hexdigest()[:8]
    tag='contextra-analytics:tracking-'+label+'-7357cdd-'+digest
    assert subprocess.run(['docker','image','inspect',tag],capture_output=True).returncode!=0
    snap=R/(label+'-snapshot');snap.mkdir()
    p=subprocess.Popen(['git','-C',str(REPO),'archive',read('plan.json')['head']],stdout=subprocess.PIPE)
    with tarfile.open(fileobj=p.stdout,mode='r|') as t:t.extractall(snap,filter='data')
    assert p.wait()==0
    for path in FILES:shutil.copyfile(REPO/path,snap/path)
    assert shutil.disk_usage('/var/lib/docker').free>=15*1024**3
    save(label+'-build-start.json',dict(hashes=hashes,tag=tag));ledger('build',tag=tag)
    started=time.monotonic()
    with (R/(label+'-build.log')).open('x') as log:
        p=subprocess.Popen(['docker','build','-f','docker/Dockerfile','--target','app','-t',tag,'.'],cwd=snap,
            env=dict(os.environ,DOCKER_BUILDKIT='1'),stdout=log,stderr=log,start_new_session=True)
        while p.poll() is None:
            if time.monotonic()-started>1200 or shutil.disk_usage('/var/lib/docker').free<15*1024**3:
                os.killpg(p.pid,signal.SIGTERM);p.wait(15);raise RuntimeError('BUILD_LIMIT')
            time.sleep(1)
    assert p.returncode==0
    image=json.loads(cmd(['docker','image','inspect',tag]))[0]['Id']
    save(label+'-build.json',dict(image=image,tag=tag,hashes=hashes,elapsed=time.monotonic()-started))
    print(label+' built '+image,flush=True);tests(label+'-image',image)
def stop(sid,label):
    ledger('stop',stream=sid,label=label);response=api('/api/v1/streams/'+sid+'/stop','POST');save(label+'-'+sid+'-stop.json',response)
    end=time.monotonic()+90
    while time.monotonic()<end:
        q=prior.state(sid)
        if q['state']=='stopped' and not q.get('cleanup_pending') and not q.get('cleanup_error'):
            save(label+'-'+sid+'-cleanup.json',q);return
        time.sleep(1)
    save(label+'-'+sid+'-stop-failed.json',q);raise RuntimeError('STOP_TIMEOUT_NO_RETRY')
def quiescent(q):
    cleanup=q.get('last_cleanup') or {}
    return (q.get('desired_state')=='stopped' and not q.get('cleanup_pending') and not q.get('cleanup_error')
        and (q['state']=='stopped' or (q['state']=='error' and cleanup.get('completed')
        and cleanup.get('generation')==q.get('run_generation')
        and q.get('metadata_path',{}).get('state')=='NOT_CONNECTED')))
def apply(label):
    assert not (R/(label+'-apply-start.json')).exists();p=read('plan.json');b=read(label+'-build.json')
    assert read(label+'-image.json')['success']
    assert sha(prior.REG)==p['registry_hash']
    allstates=api('/api/v1/streams')['streams'];assert all(x['stream_id'] in CHANNELS.values() or x['state']=='stopped' for x in allstates)
    for sid in CHANNELS.values():
        q=prior.state(sid)
        if quiescent(q):
            if q['state']=='error':save(label+'-'+sid+'-quiescent-error.json',q)
            continue
        if q['state'] in ('running','error'):
            assert not (R/(label+'-'+sid+'-stop.json')).exists(),'No repeated Stop'
            stop(sid,label)
        else:assert q['state']=='stopped'
    assert all(quiescent(prior.state(x['stream_id'])) for x in api('/api/v1/streams')['streams'])
    current=base.inspect('contextra-analytics-1');ah,eh=sha(ACTIVE),sha(ENV)
    save(label+'-apply-start.json',dict(before=current,active_hash=ah,env_hash=eh))
    shutil.copy2(ACTIVE,R/(label+'-active.before'));shutil.copy2(ENV,R/(label+'-env.before'))
    cfg=yaml.safe_load(ACTIVE.read_text());cfg['services']['analytics']['image']=b['image']
    atomic(ACTIVE,yaml.safe_dump(cfg).encode(),ah)
    lines=ENV.read_text().splitlines(keepends=True);indices=[i for i,l in enumerate(lines) if l.startswith('CONTEXTRA_IMAGE=')];assert len(indices)==1
    lines[indices[0]]='CONTEXTRA_IMAGE='+b['tag']+'\n';atomic(ENV,''.join(lines).encode(),eh)
    compose=['docker','compose','--env-file',str(ENV),'-f',str(REPO/'docker-compose.yml'),'-f',str(ACTIVE)]
    ledger('apply',label=label,image=b['image'])
    cmd(compose+['config','--quiet']);cmd(compose+['up','--no-start','--no-deps','--force-recreate','--pull','never','analytics']);cmd(['docker','start','contextra-analytics-1'])
    end=time.monotonic()+120
    while time.monotonic()<end:
        current=base.inspect('contextra-analytics-1')
        if current['State'].get('Health',{}).get('Status')=='healthy':break
        time.sleep(1)
    else:raise RuntimeError('READINESS_FAILED')
    assert current['Image']==b['image']
    assert dict(x.split('=',1) for x in current['Config']['Env'])==dict(x.split('=',1) for x in p['analytics']['Config']['Env'])
    assert base.inspect('contextra-redis-1')['Id']==p['redis']
    assert all(x['state']=='stopped' for x in api('/api/v1/streams')['streams'])
    for src,sid in CHANNELS.items():
        assert api('/api/v1/streams/'+sid+'/settings')['configured']==p['settings'][src]['configured']
        ledger('start',stream=sid,label=label);api('/api/v1/streams/'+sid+'/start','POST')
        end=time.monotonic()+90
        while time.monotonic()<end:
            q=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')
    save(label+'-applied.json',dict(image=b['image'],states={src:prior.state(sid) for src,sid in CHANNELS.items()}))
    print(label+' applied; two running',flush=True)
def observe(label):
    assert not (R/(label+'-window.json')).exists()
    time.sleep(4)
    old=prior.R;prior.R=R
    try:prior.window(label+'-window',CHANNELS)
    finally:prior.R=old
if __name__=='__main__':
    os.chdir(REPO)
    command=sys.argv[1];label=sys.argv[2] if len(sys.argv)>2 else 'probe'
    {'gate':gate,'test':lambda:tests(label),'build':lambda:build(label),'apply':lambda:apply(label),'observe':lambda:observe(label)}[command]()