from ops import *
import sys,runpy,base64
root=Path('/workspace/contextra/runtime')
rows=api('/api/v1/streams')['streams'];assert all(not r.get('cleanup_pending') and not r.get('child_pid') and not any(p.get('child_pid') for p in r.get('provider_processes',{}).values()) for r in rows)
save('pre-final-status.json',[{k:r.get(k) for k in ('stream_id','state','last_error','cleanup_error','last_cleanup')} for r in rows])
# Existing API CAS cannot remove an absent optional Fire object after materialization.
# Compare every non-trial field, then restore the original storage bytes while Runtime is stopped.
until=time.monotonic()+140
while time.monotonic()<until:
    a=api('/api/v1/metadata/status')['artifacts'];t=training().get('status',{}).get('counts',{})
    if not a['active'] and not a['ledger'].get('pending') and not a['ledger']['storage']['reserved_jobs'] and not any(t.get(k) for k in ('CAPTURING','QUEUED','RUNNING')):break
    time.sleep(2)
else:raise RuntimeError('FINAL_DRAIN_FAILED')
save('final-drained.json',dict(artifacts=a,training=training(),ledger=ledger()))
runtime=Path('/workspace/contextra/data/runtime/streams_runtime.yaml');original=PRIVATE/'runtime/streams_runtime.yaml'
code='''import json,sys,yaml
v=json.load(sys.stdin);a=yaml.safe_load(v['original']);b=yaml.safe_load(v['actual'])
assert a['removed']==b['removed'];left={x['stream_id']:x for x in a['added']};right={x['stream_id']:x for x in b['added']};assert left.keys()==right.keys()
fields={'intrusion','loitering','fall','fight','fire','crowd','tracking'}
for sid,x in left.items():
 y=right[sid]
 if str(x.get('camera_id')) not in [str(c) for c in range(30,46)]:assert x==y
 else:assert all(x.get(k)==y.get(k) for k in set(x)|set(y) if k not in fields)
print(json.dumps(dict(non_trial_fields_preserved=True,registered=len(left))))
'''
checked=json.loads(cmd(['docker','exec','-i',CONTAINER,'python3','-B','-c',code],input=json.dumps(dict(original=original.read_text(),actual=runtime.read_text())),timeout=20))
cmd(['docker','stop','--time','90',CONTAINER],100);assert inspect(CONTAINER)['State']['ExitCode']==0
shutil.copy2(runtime,PRIVATE/'runtime-post-trial.yaml')
tmp=runtime.with_suffix('.ac-restore');shutil.copy2(original,tmp);os.replace(tmp,runtime)
assert hashlib.sha256(runtime.read_bytes()).digest()==hashlib.sha256(original.read_bytes()).digest()
save('settings-restored.json',dict(**checked,exact_original_bytes=True,sha256=hashlib.sha256(runtime.read_bytes()).hexdigest()))
doc=json.loads((PRIVATE/'compose-before.json').read_text());tx=json.loads((BASE/'prepared.json').read_text())['transaction']
cmd(['docker','compose','-p','contextra-icerti','-f',str(Path(tx)/'compose.json'),'up','-d','--no-deps','--no-build','--pull','never','analytics'],150)
until=time.monotonic()+240
while time.monotonic()<until:
    try:api('/ready');break
    except Exception:time.sleep(2)
else:raise RuntimeError('FINAL_READY')
actual=inspect(CONTAINER);assert actual['Image']==doc['services']['analytics']['image']
env=dict(x.split('=',1) for x in actual['Config']['Env'])
for k,v in doc['services']['analytics']['environment'].items():assert env.get(k)==str(v),k
assert env['CONTEXTRA_TEST_NO_CAPTURE']=='false' and env['CONTEXTRA_SOURCE_LATEST_TRIAL']=='true' and env['CONTEXTRA_COMPLETED_QUEUE_MAX_AGE_SECONDS']=='1'
assert env.get('CONTEXTRA_PIPELINE_DIAGNOSTICS','false')=='false'
cmd(['docker','start','contextra-icerti-ai-gateway-1'],30)
code='''import os,json,urllib.request
from pathlib import Path
t=Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip()
r=urllib.request.Request('http://127.0.0.1:8080/internal/bridge-status',headers={'Authorization':'Bearer '+t})
print(json.dumps(json.load(urllib.request.urlopen(r,timeout=3))))'''
until=time.monotonic()+90
while time.monotonic()<until:
    bridge=json.loads(cmd(['docker','exec','contextra-icerti-ai-control-sidecar-1','python3','-B','-c',code],timeout=8))
    if bridge['auth'] is None and bridge['wss']['connected']:break
    time.sleep(3)
else:raise RuntimeError('FINAL_WSS')
rows=api('/api/v1/streams')['streams'];assert all(r['state']=='stopped' and not r.get('cleanup_pending') for r in rows)
save('final-state.json',dict(image=actual['Image'],channels_stopped=len(rows),gateway=inspect('contextra-icerti-ai-gateway-1')['State']['Running'],bridge=bridge,
    models=probe('models'),training=training(),artifact=api('/api/v1/metadata/status')['artifacts'],ledger=ledger(),
    options={k:env.get(k) for k in ('CONTEXTRA_TEST_NO_CAPTURE','CONTEXTRA_SOURCE_LATEST_TRIAL','CONTEXTRA_COMPLETED_QUEUE_MAX_AGE_SECONDS','CONTEXTRA_PROVIDER_LINEAR_TRIAL','CONTEXTRA_PROVIDER_FRESHNESS_ENABLED','CONTEXTRA_PIPELINE_DIAGNOSTICS','CONTEXTRA_INPUT_TIMING_SECONDS')},
    gpu=cmd(['nvidia-smi','--query-gpu=memory.used,memory.total','--format=csv,noheader']),
    gpu_processes=cmd(['nvidia-smi','--query-compute-apps=pid,process_name,used_memory','--format=csv,noheader'])))
print(json.dumps(dict(final=True,stopped=len(rows),wss=True)),flush=True)