import copy,json,os,subprocess,sys,time,urllib.request,uuid
from pathlib import Path
os.umask(0o077)
ROOT=Path('/backup/contextra/deployment-inbox/icerti-crowd-20260928-r01');RUNTIME=Path('/workspace/contextra/runtime')
sys.path.insert(0,str(RUNTIME/'deploy-tools/provider-20260927-r01'))
from deployment_kit import release,compose_snapshot
from deployment_kit.core import write_json,sha256
def cmd(args,timeout=25,input=None):
    p=subprocess.run(args,input=input,text=True,capture_output=True,timeout=timeout,env=compose_snapshot.command_environment())
    if p.returncode:raise RuntimeError('COMMAND_'+Path(args[0]).name+'_'+str(p.returncode))
    return p.stdout
def api(path):return json.load(urllib.request.urlopen('http://127.0.0.1:35000'+path,timeout=8))
def inspect(name):return json.loads(cmd(['docker','inspect','contextra-icerti-'+name+'-1']))[0]
def stopped():
    rows=api('/api/v1/streams')['streams']
    return rows,all(r['state']=='stopped' and not r.get('cleanup_pending') and not r.get('child_pid') and not any(v.get('child_pid') for v in r.get('provider_processes',{}).values()) for r in rows)
assert (ROOT/'runtime-restored-ready.json').exists()
rows,ok=stopped();assert ok
active=json.loads((RUNTIME/'state/active.json').read_text());p=json.loads((ROOT/'private/prepared-r02.json').read_text())
assert active['transaction']==p['transaction']
assert json.loads((RUNTIME/'state/site.json').read_text())==p['site']
doc=json.loads(Path(p['compose']).read_text())
args,hashes=release.compose_args(p['site'],RUNTIME/'releases'/active['release_id'])
current=json.loads(cmd(['docker',*args,'config','--format','json']))
def norm(d):
    d=copy.deepcopy(d)
    for s in d['services'].values():s.pop('build',None);s['volumes']=sorted(s.get('volumes',[]),key=lambda v:v['target'])
    return d
assert norm(current)==norm(doc)
manifest=json.loads((ROOT/'kit/manifest.json').read_text())
for service,base,prefix in [('analytics','/opt/contextra','app/'),('ai-control-sidecar','/opt/bridge','control_sidecar/')]:
    code="import json,hashlib;from pathlib import Path;r=Path("+repr(base)+");print(json.dumps({str(p.relative_to(r)):hashlib.sha256(p.read_bytes()).hexdigest() for p in (r/"+repr(prefix.rstrip('/'))+").rglob('*.py')}))"
    actual=json.loads(cmd(['docker','exec','contextra-icerti-'+service+'-1','python3','-B','-c',code]))
    expected={k:v for k,v in manifest['files'].items() if k.startswith(prefix) and k.endswith('.py')}
    assert actual==expected,'IMAGE_SOURCE_HASH_'+service
old_models=json.loads((ROOT/'models.json').read_text())
code="import json,urllib.request;names="+repr(list(old_models))+";print(json.dumps({m:{'ready':urllib.request.urlopen('http://localhost:8000/v2/models/'+m+'/ready',timeout=4).status,'config':json.load(urllib.request.urlopen('http://localhost:8000/v2/models/'+m+'/config',timeout=4))} for m in names}))"
assert json.loads(cmd(['docker','exec','contextra-icerti-analytics-1','python3','-B','-c',code]))==old_models
training_code="""import json,urllib.request,os
from pathlib import Path
t=Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip()
r=urllib.request.Request('http://127.0.0.1:5000/internal/secure-control/training/settings',headers={'Authorization':'Bearer '+t})
print(json.dumps(json.load(urllib.request.urlopen(r,timeout=5))))
"""
training=json.loads(cmd(['docker','exec','contextra-icerti-analytics-1','python3','-B','-c',training_code]))
cfg=training['configured']
for key,value in dict(enabled=True,minimum_persons=8,interval_seconds=900,width=1280,height=720,max_bytes=2_000_000_000_000,min_free_bytes=100_000_000_000).items():assert cfg[key]==value,key
cmd(['systemctl','start','contextra-runtime.service'],260)
cmd(['docker','exec','contextra-icerti-ai-gateway-1','nginx','-t','-c','/tmp/nginx.conf'])
rendered=cmd(['docker','exec','contextra-icerti-ai-gateway-1','cat','/tmp/nginx.conf'])
assert '(?:INTRUSION|LOITERING|FIRE|FALL|FIGHT|CROWD)' in rendered
bridge_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=4))))
"""
deadline=time.monotonic()+90
while time.monotonic()<deadline:
    bridge=json.loads(cmd(['docker','exec','contextra-icerti-ai-control-sidecar-1','python3','-B','-c',bridge_code]))
    if bridge['auth'] is None and bridge['wss']['connected']:break
    time.sleep(2)
else:raise RuntimeError('WSS_NOT_READY')
external=[]
for _ in range(3):
    time.sleep(4);rows,ok=stopped()
    if not ok:
        external=[r['stream_id'] for r in rows if r['state']!='stopped' or r.get('cleanup_pending')]
        cmd(['docker','stop','--time','10','contextra-icerti-ai-gateway-1'],20)
        for sid in external:
            request=urllib.request.Request('http://127.0.0.1:35000/api/v1/streams/'+sid+'/stop',data=b'{}',headers={'Content-Type':'application/json'})
            urllib.request.urlopen(request,timeout=20).close()
        deadline=time.monotonic()+120
        while time.monotonic()<deadline:
            rows,ok=stopped()
            if ok:break
            time.sleep(1)
        assert ok,'EXTERNAL_START_REAP'
        break
actual=inspect('analytics');env=dict(x.split('=',1) for x in actual['Config']['Env'])
for key,value in doc['services']['analytics']['environment'].items():assert env.get(key)==str(value),'ENV_DRIFT_'+key
art=api('/api/v1/metadata/status')['artifacts'];assert art['enabled'] and art['http_enabled'] and art['storage_cap_bytes']==4_000_000_000_000
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'
trial=json.loads((ROOT/'a13/trial-result.json').read_text())
runs=','.join("'"+str(uuid.UUID(r['run_id']))+"'" for r in trial['cleanup']['channels'].values())
sql="SELECT coalesce(json_agg(t),'[]') FROM (SELECT event_id,payload->>'cameraId' AS camera,payload->>'scenario' AS scenario,artifact_state,delivery_state,error,delivery_error,http_status,attempts,confirmed_at,updated_at,media->>'clip' AS clip,media->>'thumbnail' AS thumbnail FROM contextra_event_artifacts WHERE payload->>'runId' IN ("+runs+") ORDER BY confirmed_at) t;"
events=json.loads(cmd(['docker','exec','-i','contextra-icerti-postgresql-1','sh','-c','exec psql -X -qAt -U "$POSTGRES_USER" -d "$POSTGRES_DB"'],input=sql))
write_json(ROOT/'events.json',events,new=True)
state=dict(transaction=active['transaction'],images=json.loads((ROOT/'images-final.json').read_text()),channels_stopped=len(rows),
    gateway_running=inspect('ai-gateway')['State']['Running'],bridge=bridge,external_starts=external,models_ready=4,
    camera_exact_restored=json.loads((ROOT/'settings-restored.json').read_text())['exact_original_bytes'],
    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','CONTEXTRA_METADATA_STREAM_LENGTH')},
    training=training,artifact=art,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']),source_hashes_verified=True)
write_json(ROOT/'final-state.json',state,new=True)
active['acceptance'].update(DEPLOYED='PASS',MODELS='PASS',A13='MEASURED',GATEWAY='CONFIG_PASS',SETTINGS='EXACT_RESTORED',
    WSS='USER_START_HOLD' if external else 'PASS',WEB_BROWSER='USER_PENDING')
active['incremental_patch']['note']='Crowd/cooldown/Gateway patched; A13 measured once; original settings restored; manual Web acceptance pending'
write_json(RUNTIME/'state/active.json',active,expected=sha256(RUNTIME/'state/active.json'))
write_json(RUNTIME/'state/transactions'/active['transaction']/'accepted.json',active,new=True)
write_json(RUNTIME/'state/inflight.json',dict(status='DONE',stage='USER_START_HOLD' if external else 'ACCEPTED_WEB_PENDING',transaction=active['transaction']))
print(json.dumps(dict(stopped=len(rows),gateway=state['gateway_running'],wss=bool(bridge['wss']['connected']) and not external,events=len(events),external_starts=len(external))))