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

os.umask(0o077)
R = Path(__file__).parent
R.chmod(0o700)
spec = importlib.util.spec_from_file_location('ops', '/tmp/codex/ltb1021/2026-09-15_contextra-test-run_r01/ops.py')
ops = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ops)
api = ops.api
REG = Path('/data/contextra/runtime-state/streams_runtime.yaml')
SOURCES = ['rtsp-test-source-01', 'rtsp-test-source-02']

def save(name, data):
    p = R / name
    p.write_text(json.dumps(data, ensure_ascii=False, indent=2))
    p.chmod(0o600)

def read(name):
    return json.loads((R / name).read_text())

def ledger(event, **kwargs):
    p = R / 'ledger.jsonl'
    with p.open('a') as f:
        f.write(json.dumps(dict(at=time.time(), event=event, **kwargs)) + '\n')

def state(sid):
    return api('/api/v1/streams/' + sid)

def mutation(sid, action, body=None):
    ledger('request', stream=sid, action=action, body=body)
    value = api('/api/v1/streams/' + sid + '/' + action, 'POST', body)
    ledger('response', stream=sid, action=action, response=value)
    return value

def resource():
    info = ops.base.inspect('contextra-analytics-1')
    pid = info['State']['Pid']
    cg = Path('/sys/fs/cgroup' + Path(f'/proc/{pid}/cgroup').read_text().strip().split('::')[1])
    stats = dict(line.split() for line in (cg / 'cpu.stat').read_text().splitlines())
    processes, threads = {}, {}
    for raw_pid in (cg / 'cgroup.procs').read_text().split():
        try:
            p = Path('/proc') / raw_pid
            processes[raw_pid] = dict(comm=(p / 'comm').read_text().strip(), status=(p / 'status').read_text())
            for t in (p / 'task').iterdir():
                try:
                    txt = (t / 'stat').read_text()
                    fields = txt[txt.rfind(')') + 2:].split()
                    sched = (t / 'schedstat').read_text().split()
                    threads[str(t.name)] = dict(pid=int(raw_pid), comm=(t / 'comm').read_text().strip(),
                        start=fields[19], ticks=int(fields[11])+int(fields[12]),
                        runtime_ns=int(sched[0]), runqueue_wait_ns=int(sched[1]))
                except (FileNotFoundError, ProcessLookupError):
                    pass
        except (FileNotFoundError, ProcessLookupError):
            pass
    return dict(at=time.monotonic(), cgroup=str(cg), cpu={k:int(v) for k,v in stats.items()},
        memory=int((cg / 'memory.current').read_text()), processes=processes, threads=threads,
        mem_available=ops.prior.prev.mem(), gpu=ops.prior.prev.gpu(), ticks_per_second=os.sysconf('SC_CLK_TCK'))

def redis_sample(source):
    key = 'contextra:metadata:v1:camera:' + hashlib.sha256(source.encode()).hexdigest() + ':frames'
    at = time.monotonic()
    info = ops.base.m.query('reader', 'XINFO', 'STREAM', key)
    rows = ops.base.m.query('reader', 'XREVRANGE', key, '+', '-', 'COUNT', '1')
    fields = rows[0][1]
    if not isinstance(fields, dict):
        fields = dict(zip(fields[::2], fields[1::2]))
    frame = json.loads(fields['json'])
    return dict(at=at, entries=info['entries-added'], length=info['length'], last=info['last-generated-id'],
        pttl=ops.base.m.query('reader', 'PTTL', key), frame=frame)

def snapshot(channels):
    out = dict(at=time.monotonic(), channels={}, publisher=api('/api/v1/metadata/status'))
    for source, sid in channels.items():
        st = state(sid)
        assert st['state'] == 'running', (sid, st['state'])
        out['channels'][source] = dict(at=time.monotonic(), state=st, redis=redis_sample(source))
    return out

def window(name, channels):
    assert not (R / (name + '.json')).exists(), 'No repeated window'
    start_states = {source:state(sid) for source,sid in channels.items()}
    for source,sid in channels.items():
        assert not start_states[source]['metadata_path']['performance']['detail']
        mutation(sid, 'metrics', dict(run_generation=start_states[source]['run_generation'], enabled=True, seconds=45))
    samples = []
    try:
        resources = [resource()]
        samples.append(snapshot(channels))
        deadline = time.monotonic() + 18
        while time.monotonic() < deadline:
            time.sleep(min(1, max(0, deadline - time.monotonic())))
            samples.append(snapshot(channels))
            for source, value in samples[-1]['channels'].items():
                assert value['state']['run_id'] == start_states[source]['run_id'], 'Unexpected run change'
        resources.append(resource())
        assert resources[-1]['memory'] < 12*1024**3 and resources[-1]['mem_available'] > 4*1024**3, 'Resource bound reached'
        save(name + '.json', dict(samples=samples, resources=resources))
        print(name + ': captured ' + str(len(samples)) + ' bounded snapshots', flush=True)
    finally:
        if not (R / (name + '.json')).exists():
            save(name + '-partial.json', dict(samples=samples))
        for source,sid in channels.items():
            st = state(sid)
            if st['state'] == 'running' and st['run_id'] == start_states[source]['run_id']:
                mutation(sid, 'metrics', dict(run_generation=st['run_generation'], enabled=False, seconds=1))

def gate():
    assert not (R / 'plan.json').exists()
    reg = yaml.safe_load(REG.read_text())
    selected = {src: [x for x in reg['added'] if x.get('source_id') == src] for src in SOURCES}
    assert all(len(x)==1 for x in selected.values())
    registrations = {src:x[0] for src,x in selected.items()}
    channels = {src:x['stream_id'] for src,x in registrations.items()}
    states = {src:state(sid) for src,sid in channels.items()}
    assert all(x['state']=='running' for x in states.values())
    settings = {src:api('/api/v1/streams/'+sid+'/settings') for src,sid in channels.items()}
    assert all(not x['metadata_path']['performance']['detail'] for x in states.values())
    assert all(x['metadata_path'].get('visual') is None for x in states.values())
    assert all(x.get('store') is False and x.get('active_learning') is False for x in registrations.values())
    resources = resource()
    assert resources['mem_available'] > 4*1024**3 and resources['memory'] < 12*1024**3
    save('plan.json', dict(started=time.time(), channels=channels, states=states, settings=settings,
        registrations=registrations, registry_hash=ops.base.g.sha(REG), sender=ops.prior.prev.sender(),
        inventory=ops.base.inventory(), resources=resources,
        head=subprocess.check_output(['git','rev-parse','HEAD'],cwd=ops.REPO,text=True).strip(),
        config_hashes={str(p):ops.base.g.sha(p) for p in [ops.C/'compose.yaml',ops.C/'users.acl',ops.C/'redis.conf',Path('/workspace/contextra-runtime-config/runtime.env')]},
        limits=dict(windows=2,window_seconds=18,stabilize_seconds=5,stop_start_wait_seconds=90,
            run_seconds=300,task_budget_seconds=1200,snapshots_per_window=22,raw_frames=100,
            evidence_bytes=30*1024**2,helper_memory_bytes=256*1024**2,helper_cpu_seconds=30,
            analytics_memory_bytes=12*1024**3,minimum_available_ram_bytes=4*1024**3,gpu_memory_mib=16000),
        rollback='Stop01 once only. If cleanup incomplete, abort and preserve02; no container kill. After clean stop restore01 Start once. No settings or sender changes. Metrics disabled after each window, automatic expiry45s.'))
    ledger('gate_complete', channels=channels)
    print('Gate PASS: two running channels, capture/detail Off, private plan fixed', flush=True)

def alone():
    plan=read('plan.json'); channels=plan['channels']; sid=channels[SOURCES[0]]
    assert (R/'two.json').exists() and not (R/'stop-requested.json').exists()
    assert state(sid)['run_id']==plan['states'][SOURCES[0]]['run_id']
    assert ops.base.g.sha(REG)==plan['registry_hash']
    save('stop-requested.json', dict(at=time.time(),stream=sid))
    mutation(sid,'stop')
    deadline=time.monotonic()+90; history=[]; clean=False
    while time.monotonic()<deadline:
        st=state(sid);history.append(st)
        if st['state']=='stopped' and not st.get('cleanup_pending') and not st.get('cleanup_error'):
            clean=True;break
        time.sleep(1)
    save('stop-result.json',dict(clean=clean,history=history))
    if not clean:
        print('STOP_NOT_CLEAN: comparison aborted; no repeat or container stop',flush=True)
        return
    print('Channel01 normal cleanup confirmed; channel02 unchanged',flush=True)
    try:
        transition=[]
        for _ in range(5):
            transition.append(snapshot({SOURCES[1]:channels[SOURCES[1]]}));time.sleep(1)
        save('stabilization.json',transition)
        window('one',{SOURCES[1]:channels[SOURCES[1]]})
    finally:
        assert not (R/'restore-requested.json').exists()
        assert state(sid)['state']=='stopped'
        assert api('/api/v1/streams/'+sid+'/settings')['configured']==plan['settings'][SOURCES[0]]['configured']
        save('restore-requested.json',dict(at=time.time(),stream=sid))
        mutation(sid,'start')
        deadline=time.monotonic()+90; history=[]; restored=False
        while time.monotonic()<deadline:
            st=state(sid);history.append(st)
            if st['state']=='running' and st.get('metadata_path',{}).get('processed',0)>0:
                restored=True;break
            if st['state']=='error':break
            time.sleep(1)
        save('restore-result.json',dict(restored=restored,history=history))
        print('Channel01 restore: '+str(restored),flush=True)

def final():
    plan=read('plan.json')
    states={src:state(sid) for src,sid in plan['channels'].items()}
    settings={src:api('/api/v1/streams/'+sid+'/settings') for src,sid in plan['channels'].items()}
    current_hashes={p:ops.base.g.sha(Path(p)) for p in plan['config_hashes']}
    sender=ops.prior.prev.sender()
    result=dict(states=states,settings=settings,sender=sender,inventory=ops.base.inventory(),
        settings_equal={src:settings[src]['configured']==plan['settings'][src]['configured'] for src in SOURCES},
        registry_equal=ops.base.g.sha(REG)==plan['registry_hash'],
        config_equal=current_hashes==plan['config_hashes'],
        sender_execution_equal=sender['executionId']==plan['sender']['executionId'],
        channel02_run_equal=states[SOURCES[1]]['run_id']==plan['states'][SOURCES[1]]['run_id'],
        resources=resource())
    save('final.json',result)
    print(json.dumps({k:v for k,v in result.items() if k.endswith('equal')},ensure_ascii=False),flush=True)
    print(json.dumps({src:dict(state=s['state'],detail=s['metadata_path']['performance']['detail']) for src,s in states.items()}),flush=True)

if __name__=='__main__':
    mode=sys.argv[1]
    if mode=='gate':gate()
    elif mode=='two':window('two',read('plan.json')['channels'])
    elif mode=='alone':alone()
    elif mode=='final':final()