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

os.umask(0o077)
R=Path(__file__).parent
s=importlib.util.spec_from_file_location('old','/tmp/codex/ltb1021/2026-09-15_contextra-publisher_r01/ops.py')
old=importlib.util.module_from_spec(s);s.loader.exec_module(old);old.R=R;old.prev.R=R;p=old.p;p.R=R
p.FILES=['app/metadata_publisher.py','app/metadata_runtime.py','app/publisher_metrics.py','app/api/streams.py','app/application/streams.py',
         'docker-compose.yml','tests/test_publisher_metrics.py','tests/test_metadata_demo_limits.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 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,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'],stream_length=100))
    cfg=yaml.safe_load(p.ACTIVE.read_text());cfg['services']['analytics']['image']=b['image']
    cfg['services']['analytics'].setdefault('environment',{})['CONTEXTRA_METADATA_STREAM_LENGTH']='100'
    if label=='parallel':cfg['services']['analytics']['environment']['CONTEXTRA_METADATA_WORKERS']='4'
    p.atomic(p.ACTIVE,yaml.safe_dump(cfg).encode(),ah)
    lines=p.ENV.read_text().splitlines(keepends=True)
    values=[('CONTEXTRA_IMAGE',b['tag']),('CONTEXTRA_METADATA_STREAM_LENGTH','100')]
    if label=='parallel':values.append(('CONTEXTRA_METADATA_WORKERS','4'))
    for key,value in values:
        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'],stream_length=100)
    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['CONTEXTRA_METADATA_STREAM_LENGTH']='100'
    if label=='parallel':expected['CONTEXTRA_METADATA_WORKERS']='4'
    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'])
    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()}))
    print('3 channels running;stream_length100')

def observe(label):
    s=importlib.util.spec_from_file_location('w','/tmp/codex/ltb1021/2026-09-15_contextra-three-channel_r01/observe.py')
    w=importlib.util.module_from_spec(s);s.loader.exec_module(w);w.R=R;w.p.R=R
    time.sleep(5)
    p.ledger('profile',label=label,seconds=45)
    p.api('/api/v1/metadata/metrics','POST',dict(enabled=True,seconds=45))
    try:
        w.window(label+'-window',p.CHANNELS,24)
        p.save(label+'-publisher.json',p.api('/api/v1/metadata/status?include_timeline=true'))
    finally:p.api('/api/v1/metadata/metrics','POST',dict(enabled=False,seconds=1))

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