from ops import *
import sys, signal
label=sys.argv[1];assert label in ('baseline','source-baseline','source-baseline-r2','candidate')
assert all(r['state']=='stopped' and not r.get('cleanup_pending') for r in api('/api/v1/streams')['streams'])
assert not inspect('contextra-ai-gateway-1')['State']['Running']
image=inspect('contextra-analytics-1')['Image']
code=Path('/tmp/codex/ltb1021/2026-09-27_result-queue-bc_r01/drift_reader.py').read_text()
code=code.replace('(3,8,9,10)', '(3,5,6,8,9,10,11,12)').replace('20000','50000')
path=HERE/(label+'-redis.jsonl')
auth=Path('/workspace/contextra-runtime-config/metadata-active/reader/reader-credentials.json')
assert not auth.is_symlink() and auth.stat().st_mode&0o077==0
pid=None;returncode=None
with path.open('x') as log,(HERE/(label+'-reader.stderr')).open('x') as err:
    reader=subprocess.Popen(['docker','exec','-i','contextra-analytics-1','python3','-B','-c',code],stdin=subprocess.PIPE,stdout=log,stderr=err,text=True)
    try:
        reader.stdin.write(auth.read_text());reader.stdin.close();deadline=time.monotonic()+10
        while True:
            with path.open() as stream:first=stream.readline()
            if first:
                ready=json.loads(first);assert ready['kind']=='ready';pid=ready['pid'];break
            if reader.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Reader startup failed')
            time.sleep(.1)
        print('Read-only RTP collector ready; '+label+' trial begins',flush=True)
        trial=subprocess.run([sys.executable,str(HERE/'trial.py'),label],timeout=1050)
        returncode=trial.returncode
    finally:
        if pid and reader.poll() is None:
            assert inspect('contextra-analytics-1')['Image']==image
            kill="import os,signal;from pathlib import Path;p="+str(pid)+";a=Path('/proc')/str(p)/'cmdline';assert a.exists() and b'contextra-drift-live-r01' in a.read_bytes();os.kill(p,signal.SIGTERM)"
            cmd(['docker','exec','contextra-analytics-1','python3','-B','-c',kill],timeout=5)
        reader.wait(timeout=10)
save(label+'-reader-final.json',dict(reader_exit=reader.returncode,trial_exit=returncode,pid=pid,reaped=True))
raise SystemExit(returncode or reader.returncode)