import copy,csv,hashlib,json,os,shutil,subprocess,sys,time
from pathlib import Path
from urllib.parse import quote
from urllib.request import build_opener,ProxyHandler
import yaml
os.umask(0o077)
R=Path(__file__).parent;R.chmod(0o700)
REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor');sys.path.insert(0,str(REPO))
from tools.stream_benchmark import load_config
from tools.stream_benchmark_runtime import Runtime,file_hash,subset
from tools.stream_benchmark_metrics import summarize
from tools.test_run_manager import load_preset
CFG=Path('/workspace/contextra-runtime-config/test-run/event-artifact-next.yaml')
ACTIVE=Path('/workspace/contextra-runtime-config/metadata-active/compose.yaml')
ENV=Path('/workspace/contextra-runtime-config/runtime.env')
REG=Path('/data/contextra/runtime-state/streams_runtime.yaml')
MANIFEST=Path('/workspace/contextra-runtime-config/test-run/medium16-registry.json')
ROOT=Path('/data/contextra/artifacts')
TARGET={'web-cam08-test':2,'web-cam01-test':1}
TAG='contextra-analytics:event-artifact-c30a1db-76cb7089'
def save(name,value):
 p=R/name;p.write_text(json.dumps(value,ensure_ascii=False,indent=2));p.chmod(0o600)
def read(name):return json.loads((R/name).read_text())
def cmd(args,timeout=30):
 p=subprocess.run(args,capture_output=True,text=True,timeout=timeout)
 if p.returncode:
  save('command-error-'+str(time.time_ns())+'.json',dict(args=args,code=p.returncode,stdout=p.stdout[-4000:],stderr=p.stderr[-4000:]))
  raise RuntimeError('Command failed; see private evidence')
 return p.stdout
def runtime():return Runtime(load_config(CFG),R)
def ledger():
 sql="SELECT coalesce(json_agg(x),'[]') FROM (SELECT * FROM contextra_event_artifacts ORDER BY created_at) x"
 return json.loads(cmd(['docker','exec','contextra-postgresql-1','sh','-c','psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -At -c "$1"','sh',sql]))
def gate():
 assert not (R/'plan.json').exists()
 rt=runtime();inspect=rt.inspect();assert inspect['State']['Status']=='exited' and inspect['State']['Pid']==0
 cfg=load_config(CFG);image=json.loads(cmd(['docker','image','inspect',TAG]))[0];assert image['Id']==cfg['expected_image']
 prior=json.loads(Path('/tmp/codex/ltb1021/2026-09-17_contextra-event-artifact_r01/final-image.json').read_text())
 assert image['Id']==prior['id']
 for p,h in prior['hashes'].items():
  if p.startswith(('app/','native/','tools/','tests/')):assert file_hash(REPO/p)==h
 spec=yaml.safe_load(REG.read_text());assert not any(str(s.get('source_id')) in ['1','2'] for s in spec['added'])
 old={s['stream_id']:s for s in spec['added']};assert all(not s.get('autostart') for s in spec['added'])
 assert old['web-cam08-test']['camera_id']==77 and old['web-cam01-test']['camera_id']==78
 assert all(not old[s]['store'] and not old[s]['active_learning'] for s in TARGET)
 assert shutil.disk_usage('/var/lib/docker').free>15*1024**3 and shutil.disk_usage(ROOT).free>60*1024**3
 entries=ledger();assert all(e['delivery_state'] in ['FAILED','SENT','BLOCKED_TIME'] for e in entries)
 presets=[Path('/workspace/contextra-runtime-config/test-run/presets')/n for n in ['web-cam08.yaml','web-cam01.yaml']]
 for f,label in [(ACTIVE,'compose'),(ENV,'env'),(REG,'registry'),(MANIFEST,'manifest'),(CFG,'config')]:shutil.copy2(f,R/(label+'.before'))
 p=dict(head=cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip(),started=time.time(),inspect=inspect,image=image['Id'],
  active_hash=file_hash(ACTIVE),env_hash=file_hash(ENV),registry=spec,registry_hash=file_hash(REG),manifest=json.loads(MANIFEST.read_text()),
  presets={str(f):dict(hash=file_hash(f),value=yaml.safe_load(f.read_text())) for f in presets},config=cfg,
  previous_events=entries,redis_id=json.loads(cmd(['docker','inspect','contextra-redis-1']))[0]['Id'],
  postgres_id=json.loads(cmd(['docker','inspect','contextra-postgresql-1']))[0]['Id'],
  caps=dict(warmup_seconds=5,window_seconds=60,wall_seconds=100,frames_per_channel=1200,
   container_ram_gib=16,host_available_gib=8,free_disk_gib=15,gpu_mib=40000,artifact_pending=64),failure='STOP_ONLY_KEEP_NEW_IDS_AND_IMAGE')
 save('plan.json',p);print('Gate recorded: stopped, IDs1/2 free, source/image match, historical events terminal',flush=True)
def reflect():
 p=read('plan.json');assert not (R/'reflect-intent.json').exists()
 assert file_hash(ACTIVE)==p['active_hash'] and file_hash(ENV)==p['env_hash'] and file_hash(REG)==p['registry_hash']
 a=yaml.safe_load(ACTIVE.read_text());a['services']['analytics']['image']=p['image']
 save('reflect-intent.json',dict(at=time.time(),image=p['image'],before_hash=p['active_hash']))
 temp=ACTIVE.with_suffix('.pending');temp.write_text(yaml.safe_dump(a,sort_keys=False));temp.chmod(0o600);os.replace(temp,ACTIVE)
 cmd(['docker','compose','-p','contextra','--env-file',str(ENV),'-f',str(REPO/'docker-compose.yml'),'-f',str(ACTIVE),
      'up','-d','--no-deps','--no-build','--pull','never','analytics'],180)
 rt=runtime();deadline=time.monotonic()+120
 while time.monotonic()<deadline:
  try:
   states=rt.states()
   if states:break
  except Exception:pass
  time.sleep(2)
 else:raise RuntimeError('Analytics readiness timeout')
 assert all(q['state']=='stopped' and not q.get('child_pid') for q in states.values())
 assert rt.inspect()['Image']==p['image']
 save('reflected.json',dict(states=states,active_hash=file_hash(ACTIVE),settings={s:rt.settings(s) for s in TARGET},resources=rt.resources()))
 print('Prepared image reflected; no channel started',flush=True)
def migrate():
 p=read('plan.json');rt=runtime();old={s['stream_id']:s for s in p['registry']['added']}
 assert not (R/'migrate-intent.json').exists();save('migrate-intent.json',dict(mapping=TARGET,settings=read('reflected.json')['settings']))
 allowed=['server_id','source_type','uri','site_id','store','active_learning','autostart','optional_roles','inference_profile','frame_time_anchor','source_options','persist_every','persist_mode','scenarios','tracking','intrusion']
 for sid,ident in TARGET.items():
  q=rt.states()[sid];assert q['state']=='stopped' and not q.get('child_pid') and not q.get('cleanup_pending')
  body={k:v for k,v in old[sid].items() if k in allowed};body.update(stream_id=sid,source_id=str(ident),camera_id=ident,autostart=False,analysisFpsTarget=10)
  conf=read('reflected.json')['settings'][sid]
  settingkeys=['tracking','pose','fire','intrusion','display_score','downstream_score','display_categories','debug_all_candidates','frame_time_anchor','analysisFpsTarget']
  changes={k:v for k,v in conf.items() if k in settingkeys}
  save('delete-intent-'+sid+'.json',dict(state=q))
  rt.api.call('DELETE','/api/v1/streams/'+quote(sid,safe=''))
  save('register-intent-'+sid+'.json',body);rt.api.call('POST','/api/v1/streams?start=false',body)
  patch=dict(expected_revision=rt.settings(sid)['revision'],**changes);save('patch-intent-'+sid+'.json',patch)
  rt.api.call('PATCH','/api/v1/streams/'+quote(sid,safe='')+'/settings',patch)
  assert subset(changes,rt.settings(sid))
  for name,data in p['presets'].items():
   if data['value']['camera']['stream_id']!=sid:continue
   file=Path(name);assert file_hash(file)==data['hash'];preset=copy.deepcopy(data['value']);preset['camera']['source_id']=ident
   file.write_text(yaml.safe_dump(preset,allow_unicode=True,sort_keys=False));file.chmod(0o600)
   loaded=load_preset(file);assert loaded['camera']['source_id']==str(ident) and loaded['camera_id']==ident
  save('migrated-'+sid+'.json',dict(settings=rt.settings(sid),state=rt.states()[sid]))
 manifest=copy.deepcopy(p['manifest'])
 for row in manifest['channels']:
  if row['stream_id'] in TARGET:row.update(source_id=str(TARGET[row['stream_id']]),camera_id=TARGET[row['stream_id']])
 assert MANIFEST.read_bytes()==(R/'manifest.before').read_bytes();MANIFEST.write_text(json.dumps(manifest,ensure_ascii=False,indent=2));MANIFEST.chmod(0o600)
 after=yaml.safe_load(REG.read_text());current={s['stream_id']:s for s in after['added']}
 for sid,spec in old.items():
  if sid not in TARGET:assert current[sid]==spec
 for sid,ident in TARGET.items():assert current[sid]['camera_id']==ident and current[sid]['source_id']==str(ident)
 cfg=load_config(CFG)
 for row in cfg['artifacts']:
  if row['path']==str(REG):row['sha256']=file_hash(REG)
 cfg['expected_image']=p['image']
 for file in [CFG,CFG.with_name('event-artifact.yaml')]:
  file.write_text(yaml.safe_dump(cfg,sort_keys=False));file.chmod(0o600)
 save('migrated.json',dict(registry_hash=file_hash(REG),rows=[r for r in manifest['channels'] if r['stream_id'] in TARGET],states=rt.states(),config_hash=file_hash(CFG)))
 print('IDs permanently updated: cam08=2, cam01=1; other registrations preserved',flush=True)
def fresh_events():
 previous={e['event_id'] for e in read('plan.json')['previous_events']}
 return [e for e in ledger() if e['event_id'] not in previous]
def stop_all():
 rt=runtime();result={}
 try:
  states=rt.states()
  for sid in TARGET:
   if states[sid].get('child_pid') or states[sid]['state'] in ['running','starting']:
    try:result[sid]=rt.stop(sid)
    except Exception as exc:result[sid]=dict(error=type(exc).__name__,actual=rt.states().get(sid))
  save('cleanup.json',dict(targets=result,states=rt.states(),events=fresh_events()))
 finally:
  if rt.inspect()['State']['Running'] and not (R/'container-stop-intent.json').exists():
   save('container-stop-intent.json',dict(at=time.time(),container=rt.inspect()['Id']))
   cmd(['docker','stop','-t','80',rt.container],95)
  save('container-stop.json',rt.inspect()['State'])
def get_media(events):
 receipts=[]
 for ident in TARGET.values():
  event=next((e for e in reversed(events) if e['payload']['cameraId']==ident and e['artifact_state']=='READY'),None)
  if event is None:continue
  body=(event['body'] or {}).get('events',[{}])[0]
  for key,filename in [('clipUrl','clip.mp4'),('thumbnailUrl','thumbnail.jpg')]:
   url=body.get(key)
   if not url:continue
   with build_opener(ProxyHandler({})).open(url,timeout=5) as response:
    payload=response.read(16*1024**2);digest=hashlib.sha256(payload).hexdigest()
    assert digest==file_hash(ROOT/event['relative_path']/filename)
    receipts.append(dict(cameraId=ident,eventId=event['event_id'],file=filename,url=url,status=response.status,size=len(payload),sha256=digest))
 save('http-get.json',receipts);return receipts
def trial():
 rt=runtime();rows=read('migrated.json')['rows'];assert not (R/'trial-intent.json').exists();save('trial-intent.json',dict(at=time.time(),caps=read('plan.json')['caps']))
 samples=[];failure=None;complete=False
 try:
  for row in rows:save('first-frame-'+row['stream_id']+'.json',rt.start(row['stream_id']))
  time.sleep(5);begin=time.monotonic()
  while True:
   sample=rt.sample(rows);sample['events']=fresh_events();sample['states']=rt.states();samples.append(sample);save('window.json',samples)
   for e in sample['events']:
    if e['artifact_state']=='FAILED' or e['delivery_state'] in ['FAILED','BLOCKED_TIME']:raise RuntimeError('Event delivery/artifact boundary failed')
   for sid,q in sample['channels'].items():
    if q['error'] or q['state']!='running' or q['processed']>1200:raise RuntimeError('Runtime failed or Frame cap reached')
    art=sample['states'][sid]['metadata_path']['artifacts']
    if art['error'] or art.get('unaccepted_events',0):raise RuntimeError('Capture admission failed')
    first=samples[0]['channels'][sid]
    if q['stale']>first['stale'] or q['drop']>first['drop'] or q['queue']>8:raise RuntimeError('Result queue/stale/drop deteriorated')
   if time.monotonic()-begin>=60:complete=True;break
   time.sleep(min(5,max(0,60-(time.monotonic()-begin))))
  summary=summarize(samples,complete);save('summary.json',summary)
  with (R/'summary.csv').open('w') as f:
   w=csv.DictWriter(f,fieldnames=list(summary['rows'][0]));w.writeheader();w.writerows(summary['rows'])
  media=get_media(fresh_events())
  if len(media)!=4:raise RuntimeError('Two-channel media GET not observed')
  if not all(any(e['payload']['cameraId']==i and e['delivery_state']=='SENT' for e in fresh_events()) for i in TARGET.values()):raise RuntimeError('Web acceptance not observed for both cameras')
 except Exception as exc:
  failure=str(exc);save('failure.json',dict(error=failure,states=rt.states(),events=fresh_events()))
  try:get_media(fresh_events())
  except Exception as ex:save('get-failure.json',dict(error=type(ex).__name__))
  stop_all()
 finally:
  save('final.json',dict(failure=failure,complete_window=complete,events=fresh_events(),registry=yaml.safe_load(REG.read_text()),inspect=rt.inspect(),
    states=rt.states() if rt.inspect()['State']['Running'] else read('cleanup.json')['states']))
  if len(samples)>1 and not (R/'summary.json').exists():save('summary.json',summarize(samples,False))
 print('Trial finished: '+str(failure or 'Web accepted; two channels running'),flush=True)
def audit():
 p=read('plan.json');final=read('final.json');now=ledger();old={e['event_id']:e for e in p['previous_events']};all_now={e['event_id']:e for e in now}
 assert all(all_now[key]==e for key,e in old.items())
 schema=[]
 root_allowed={'serverId','events'}
 item_allowed={'eventId','cameraId','recServerId','recCameraId','eventType','detectedAtWallMs','startWallTimeMs','rtpTimestamp','score','severityCode','clipUrl','thumbnailUrl','detail'}
 for e in final['events']:
  if not e['body']:continue
  body=e['body'];assert len(body['events'])==1;v=body['events'][0]
  check=dict(event_id=e['event_id'],root_extra=sorted(set(body)-root_allowed),event_extra=sorted(set(v)-item_allowed),
   camera_id_type=type(v['cameraId']).__name__,server_id=body.get('serverId'),response=e['response'],http=e['http_status'],
   types_valid=isinstance(body.get('serverId'),str) and type(v['cameraId']) is int and v['cameraId'] in (1,2)
   and type(v['detectedAtWallMs']) is int and v['detectedAtWallMs']>0 and isinstance(v['eventId'],str) and len(v['eventId'])<=100
   and v['eventType']=='INTRUSION' and type(v['score']) in (int,float) and 0<=v['score']<=1
   and v['severityCode'] in ['INTEREST','CAUTION','WARNING','DANGER'] and isinstance(v['detail'],dict)
   and all(type(v[k]) is int for k in ['startWallTimeMs','rtpTimestamp'])
   and all(isinstance(v[k],str) and len(v[k])<=2048 for k in ['clipUrl','thumbnailUrl']))
  schema.append(check)
 save('schema-comparison.json',schema)
 actual=yaml.safe_load(REG.read_text());byid={v['stream_id']:v for v in actual['added']};original={v['stream_id']:v for v in p['registry']['added']}
 assert all(byid[sid]==spec for sid,spec in original.items() if sid not in TARGET)
 rows=[];samples=read('window.json');last=samples[-1]
 for sid,ident in TARGET.items():
  assert byid[sid]['source_id']==str(ident) and byid[sid]['camera_id']==ident and not byid[sid]['autostart']
  first=read('first-frame-'+sid+'.json');end=last['states'][sid];span=(end['child_sampled_ns']-first['child_sampled_ns'])/1e9
  m=last['channels'][sid];events=[e for e in final['events'] if e['payload']['cameraId']==ident]
  rows.append(dict(stream_id=sid,cameraId=ident,startup_reference_seconds=span,
    startup_reference_result_fps=(end['metadata_path']['processed']-first['metadata_path']['processed'])/span,
    result_fps=None,redis_fps=None,queue_observed=m['queue'],stale_observed=m['stale'],drop_observed=m['drop'],
    events=len(events),artifact_ready=sum(e['artifact_state']=='READY' for e in events),
    web_failed=sum(e['delivery_state']=='FAILED' for e in events),unattempted=sum(e['attempts']==0 for e in events)))
  for filename,data in p['presets'].items():
   if data['value']['camera']['stream_id']==sid:
    spec=load_preset(filename);assert spec['camera']['source_id']==str(ident) and spec['camera_id']==ident and spec['test_label']==data['value']['test_label']
 a=yaml.safe_load(ACTIVE.read_text());assert a['services']['analytics']['image']==p['image']==final['inspect']['Image']
 assert json.loads(cmd(['docker','inspect','contextra-redis-1']))[0]['Id']==p['redis_id']
 assert json.loads(cmd(['docker','inspect','contextra-postgresql-1']))[0]['Id']==p['postgres_id']
 sender=json.loads(cmd(['python3',str(REPO/'tools/rtsp_test.py'),'list']));assert all(x['state']=='STOPPED' for x in sender if x['name']=='sample01')
 save('summary.json',dict(rows=rows,steady_window_completed=False,web='HTTP404_SERVER_NOT_FOUND',
  runtime_image=p['image'],old_events_unchanged=True,other_registrations_unchanged=True,source_changes=False,
  resource_sample=last['resource'],response=next(e['response'] for e in final['events'] if e['response']),
  stopped=final['inspect']['State']['Status'],sender=sender,unattempted_events=sum(e['attempts']==0 for e in final['events'])))
 with (R/'summary.csv').open('w') as f:
  w=csv.DictWriter(f,fieldnames=list(rows[0]));w.writeheader();w.writerows(rows)
 print('Final audit PASS: IDs persistent, old events/other registrations unchanged, no schema extras; Web SERVER_NOT_FOUND',flush=True)
if __name__=='__main__':globals()[sys.argv[1]]()