import csv
import hashlib
import importlib.util
import json
from pathlib import Path
import subprocess

R=Path(__file__).parent
s=importlib.util.spec_from_file_location('ops',R/'ops.py');o=importlib.util.module_from_spec(s);s.loader.exec_module(o);p=o.previous
plan=p.read('plan.json');built=p.read('active-build.json');data=p.read('active-window.json')
before=json.loads(Path('/tmp/codex/ltb1021/2026-09-15_contextra-tracking-fix_r01/fix-window.json').read_text())
rows=[]
for label,window in [('before',before),('after',data)]:
    for src in p.CHANNELS:
        samples=[s['channels'][src] for s in window['samples']];a,b=samples[0],samples[-1];dt=b['at']-a['at']
        am,bm=a['state']['metadata_path'],b['state']['metadata_path'];du=bm['performance']['durations'];q=[s['state']['metadata_path']['queued'] for s in samples]
        row=dict(condition=label,source=src,seconds=dt)
        for stage in ['decoded','selected','infer','native','result','publisher_submit']:
            row[stage+'_fps']=(bm['performance']['counts'].get(stage,0)-am['performance']['counts'].get(stage,0))/dt
        row['redis_fps']=(b['redis']['entries']-a['redis']['entries'])/(b['redis']['at']-a['redis']['at'])
        row.update(queue_start=q[0],queue_end=q[-1],queue_min=min(q),queue_max=max(q))
        for key in ['stale','queue_drop','publisher_rejected']:row[key+'_delta']=bm.get(key,0)-am.get(key,0)
        for key in ['tracking','tracking_cpu','scenario','result_work','queue']:
            for stat in ['mean_ms','p95_ms','max_ms']:row[key+'_'+stat]=du[key][stat]
        row['publisher_queue_max']=max(s['publisher']['metrics']['queued'] for s in window['samples'])
        row['redis_len_min']=min(s['redis']['length'] for s in samples);row['redis_len_max']=max(s['redis']['length'] for s in samples)
        row['redis_pttl_min']=min(s['redis']['pttl'] for s in samples);row['redis_pttl_max']=max(s['redis']['pttl'] for s in samples)
        rows.append(row)
with (R/'comparison.csv').open('w',newline='') as f:
    w=csv.DictWriter(f,fieldnames=list(rows[0]));w.writeheader();w.writerows(rows)
current=p.base.inspect('contextra-analytics-1')
assert current['Image']==built['image'] and current['State']['Health']['Status']=='healthy'
assert current['HostConfig']['RestartPolicy']==plan['analytics']['HostConfig']['RestartPolicy']
assert sorted(current['Mounts'],key=lambda m:m['Destination'])==sorted(plan['analytics']['Mounts'],key=lambda m:m['Destination'])
assert dict(x.split('=',1) for x in current['Config']['Env'])==dict(x.split('=',1) for x in plan['analytics']['Config']['Env'])
assert p.sha(p.prior.REG)==plan['registry_hash']
sender=p.prior.ops.prior.prev.sender();assert sender['executionId']==plan['sender']['executionId']
assert p.base.inspect('contextra-redis-1')['Id']==plan['redis']
states={src:p.prior.state(sid) for src,sid in p.CHANNELS.items()}
settings={src:p.api('/api/v1/streams/'+sid+'/settings') for src,sid in p.CHANNELS.items()}
assert all(settings[src]['configured']==plan['settings'][src]['configured'] for src in states)
assert all(q['state']=='running' and q['last_error'] is None and not q['metadata_path']['performance']['detail'] for q in states.values())
def others(inv):return {k:v for k,v in inv.items() if 'contextra-analytics-1' not in k and 'contextra-analytics-1' not in str(v)}
assert others(p.base.inventory())==others(plan['inventory'])
hash_output=subprocess.check_output(['docker','exec','contextra-analytics-1','sha256sum','/opt/contextra/app/byte_tracking.py','/opt/contextra/app/tracking_kernel.py'],text=True)
assert all(h in hash_output and p.sha(p.REPO/name)==h for name,h in built['hashes'].items() if name.startswith('app/'))
events=set();checks=[]
for src in states:
    frames=[s['channels'][src]['redis']['frame'] for s in data['samples']]
    assert len({f['runId'] for f in frames})==1
    assert all(f['sourceId']==src and f['frameTime']['rtpClockRate']==90000 and f['frameTime']['syncStatus']=='ESTIMATED' for f in frames)
    error=max(abs(f['frameTime']['detectedAtWallMs']-f['frameTime']['startWallTimeMs']-f['frameTime']['rtpTimestamp']/90) for f in frames)
    assert error<2
    for f in frames:
        for event in f.get('events',[]):events.add(event['eventId'])
    checks.append(dict(source=src,frames=len(frames),anchor_math_error_ms=error,objects_min=min(len(f['objects']) for f in frames),objects_max=max(len(f['objects']) for f in frames)))
summary=dict(status='TWO_CHANNEL_TARGET_OBSERVED',rows=rows,image=built,states=states,time_checks=checks,
    events_in_sparse_samples=len(events),settings_preserved=True,sender_preserved=True,other_services_preserved=True,
    actual_persistent_image_aligned=True,restart=current['HostConfig']['RestartPolicy'],hashes=hash_output,
    resources=data['resources'],limitations=['Short two-channel window; not long-running or16-channel capacity proof.',
    'Historical replay is full archived candidates, not the live tracker state.',
    'Measured contention reduction does not identify all delay as GIL time.',
    'Transient spikes persisted; queues recovered and no stale/drop increase was observed.'])
p.save('summary.json',summary)
print(json.dumps(dict(status=summary['status'],states={k:v['state'] for k,v in states.items()},image=built['tag'],checks=checks,events=len(events))))