import copy, hashlib, json, os, shutil, subprocess, sys, time, tarfile, signal
from pathlib import Path
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_runtime import Runtime,file_hash
from tools.stream_benchmark import load_config
CFG=Path('/workspace/contextra-runtime-config/test-run/fire8-fp16-tracking.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')
ROOT=Path('/data/contextra/artifacts')
def save(n,v):
 p=R/n;p.parent.mkdir(parents=True,exist_ok=True);p.write_text(json.dumps(v,ensure_ascii=False,indent=2));p.chmod(0o600)
def read(n):return json.loads((R/n).read_text())
def cmd(a,t=30):
 p=subprocess.run(a,capture_output=True,text=True,timeout=t)
 if p.returncode:
  save('error-'+str(time.time_ns())+'.json',dict(command=a,returncode=p.returncode,stdout=p.stdout[-1000:],stderr=p.stderr[-4000:]))
  raise RuntimeError('Command failed; see private evidence')
 return p.stdout
def gate():
 assert not (R/'plan.json').exists()
 rt=Runtime(load_config(CFG),R);states=rt.states();specs=yaml.safe_load(REG.read_text())
 manifest=json.loads(Path(rt.config['identity_manifest']).read_text());rows=manifest['channels'];assert len(rows)==16
 for p,name in [(ACTIVE,'compose.before'),(ENV,'env.before'),(REG,'registry.before')]:shutil.copy2(p,R/name)
 presets=Path('/workspace/contextra-runtime-config/test-run/presets')
 p=dict(started=time.time(),head=cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip(),config=rt.config,
        states=states,rows=rows,specs=specs,settings={r['stream_id']:rt.settings(r['stream_id']) for r in rows},
        inspect=rt.inspect(),active_hash=file_hash(ACTIVE),env_hash=file_hash(ENV),registry_hash=file_hash(REG),
        files={str(f):file_hash(f) for f in presets.rglob('*.yaml')},
        models={n:rt.triton('v2/models/'+n+'/config') for n in ['RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke']},
        sender=json.loads(cmd(['python3',str(REPO/'tools/rtsp_test.py'),'list'])),
        redis=json.loads(cmd(['docker','inspect','contextra-redis-1']))[0]['Id'],
        pg=json.loads(cmd(['docker','inspect','contextra-postgresql-1']))[0]['Id'],
        resources=rt.resources(),caps=dict(build_seconds=900,free_disk_gib=15,artifact_cap_gib=100,
        artifact_free_low_gib=50,artifact_free_resume_gib=60,trial_seconds=90,trial_frames_per_channel=1000,
        container_ram_gib=16,gpu_mib=40000,helper_workers=1,artifact_job_seconds=60),rollback='STOP_ONLY_NO_AUTOMATIC_RESTORE')
 model_files={}
 for m in p['inspect']['Mounts']:
  if m['Destination'].startswith('/opt/contextra/models/triton/'):
   model_files.update({str(f):file_hash(f) for f in Path(m['Source']).rglob('*') if f.is_file()})
 p['model_files']=model_files
 save('plan.json',p);print('Gate recorded; three running, sixteen owned registrations; no mutation')
def build(name='image'):
 assert not (R/(name+'-start.json')).exists()
 assert shutil.disk_usage('/var/lib/docker').free>15*1024**3
 paths=cmd(['git','-C',str(REPO),'diff','--name-only']).splitlines()+cmd(['git','-C',str(REPO),'ls-files','--others','--exclude-standard']).splitlines()
 hashes={p:file_hash(REPO/p) for p in paths};digest=hashlib.sha256(json.dumps(hashes,sort_keys=True).encode()).hexdigest()[:8]
 tag='contextra-analytics:event-artifact-c30a1db-'+digest
 snapshot=R/('snapshot' if name=='image' else 'snapshot-'+name);snapshot.mkdir()
 child=subprocess.Popen(['git','-C',str(REPO),'archive',read('plan.json')['head']],stdout=subprocess.PIPE)
 with tarfile.open(fileobj=child.stdout,mode='r|') as t:t.extractall(snapshot,filter='data')
 assert child.wait()==0
 for p in paths:
  (snapshot/p).parent.mkdir(parents=True,exist_ok=True);shutil.copy2(REPO/p,snapshot/p)
 save('snapshot.json' if name=='image' else 'snapshot-'+name+'.json',dict(tag=tag,hashes=hashes))
 args=['docker','build','--pull=false','--network=default','-f','docker/Dockerfile','--target','app','-t',tag,'.']
 save(name+'-start.json',dict(command=args,started=time.time()))
 with (R/(name+'.log')).open('x') as log:
  p=subprocess.Popen(args,cwd=snapshot,stdout=log,stderr=log,start_new_session=True);started=time.monotonic()
  while p.poll() is None:
   if time.monotonic()-started>900 or shutil.disk_usage('/var/lib/docker').free<15*1024**3:
    os.killpg(p.pid,signal.SIGTERM);p.wait(20);raise RuntimeError('Build bound reached; no retry/rollback')
   time.sleep(2)
 assert p.returncode==0,'Image failed; evidence retained'
 save(name+'.json',dict(tag=tag,id=json.loads(cmd(['docker','image','inspect',tag]))[0]['Id'],hashes=hashes,elapsed=time.monotonic()-started))
 print('Image ready',tag,flush=True)
def stop_original():
 rt=Runtime(load_config(CFG),R)
 for row in read('plan.json')['rows']:
  sid=row['stream_id']
  if (R/('old-stop-'+sid+'.json')).exists():continue
  q=rt.states()[sid]
  if q['state']!='stopped' or q.get('child_pid'):
   save('old-stop-'+sid+'.json',rt.stop(sid))
 save('old-all-stopped.json',rt.states())
 sender=json.loads(cmd(['python3',str(REPO/'tools/rtsp_test.py'),'list']))
 old=read('plan.json')['sender'][0]
 actual=[s for s in sender if s['name']=='sample01'];assert len(actual)==1 and actual[0]['executionId']==old['executionId']
 if actual[0]['state']!='STOPPED':
  assert not (R/'sender-stop-start.json').exists();save('sender-stop-start.json',dict(name='sample01',execution=old['executionId']))
  save('sender-stopped.json',json.loads(cmd(['python3',str(REPO/'tools/rtsp_test.py'),'stop','sample01'],60)))
 print('Trial channels stopped/reaped; owned sample01 sender stopped',flush=True)
def reflect():
 p=read('plan.json');image=read('image.json');assert not (R/'reflect-start.json').exists()
 assert file_hash(ACTIVE)==p['active_hash'] and file_hash(ENV)==p['env_hash']
 rt=Runtime(load_config(CFG),R);states=rt.states()
 assert all(states[r['stream_id']]['state']=='stopped' and not states[r['stream_id']].get('child_pid') for r in p['rows'])
 import socket
 with socket.socket() as s:s.bind(('192.168.50.221',35001))
 ROOT.mkdir(mode=0o750,parents=True,exist_ok=True)
 shutil.copy2(REPO/'docs/artifacts/README.ko.md',ROOT/'README.ko.md')
 a=yaml.safe_load(ACTIVE.read_text());v=a['services']['analytics'];v['image']=image['id']
 v['environment'].update(CONTEXTRA_EVENT_ARTIFACT_ENABLED='1',CONTEXTRA_ARTIFACT_ROOT='/data/contextra/artifacts',
  CONTEXTRA_EVENT_ENDPOINT='http://192.168.50.19:9090/api/v1/events',CONTEXTRA_ARTIFACT_PUBLIC_BASE='http://192.168.50.221:35001/artifacts')
 v.setdefault('volumes',[]).append(dict(type='bind',source=str(ROOT),target='/data/contextra/artifacts',bind={'create_host_path':False}))
 v['ports']=[dict(target=5001,published='35001',host_ip='192.168.50.221',protocol='tcp')]
 save('reflect-start.json',dict(image=image['id'],old_image=p['inspect']['Image'],old_compose_hash=p['active_hash'],old_env_hash=p['env_hash']))
 tmp=ACTIVE.with_suffix('.pending');tmp.write_text(yaml.safe_dump(a,sort_keys=False));tmp.chmod(0o600);assert file_hash(ACTIVE)==p['active_hash'];os.replace(tmp,ACTIVE)
 argv=['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']
 save('reflect-command.json',dict(command=argv));cmd(argv,180)
 deadline=time.monotonic()+120
 while time.monotonic()<deadline:
  try:
   q=rt.api.call('GET','/ready')
   if q:break
  except Exception:pass
  time.sleep(2)
 else:raise RuntimeError('Analytics readiness deadline')
 actual=rt.inspect();assert actual['Image']==image['id']
 save('reflected.json',dict(inspect=actual,active_hash=file_hash(ACTIVE),states=rt.states()))
 print('Image/config reflected; all targets remain stopped',flush=True)
def migrate():
 from urllib.parse import quote
 from tools.test_run_manager import load_preset
 p=read('plan.json');rt=Runtime(load_config(CFG),R)
 assert (R/'reflected.json').exists() and not (R/'migration-plan.json').exists()
 old={s['stream_id']:s for s in p['specs']['added']};target=[r['stream_id'] for r in p['rows']]
 original=['web-cam08-test','web-cam01-test'];order=original+[s for s in target if s not in original]
 allids={s.get('source_id') for sid,s in old.items() if sid not in target}
 assert not (allids&set(map(str,range(77,93))))
 presetroot=Path('/workspace/contextra-runtime-config/test-run/presets')
 bases=[yaml.safe_load((presetroot/n).read_text()) for n in ['web-cam08.yaml','web-cam01.yaml']]
 rows=[];plans=[];copies=[0,0]
 for i,sid in enumerate(order):
  group=i%2;origin=original[group];ident=77+i;copies[group]+=1
  base=copy.deepcopy(bases[group]);spec=copy.deepcopy(old[origin]);settings=copy.deepcopy(p['settings'][origin])
  settings={k:v for k,v in settings.items() if k in ['tracking','pose','fire','intrusion','display_score','downstream_score','display_categories','debug_all_candidates','frame_time_anchor','analysisFpsTarget']}
  settings['fire']={'enabled':False};settings['pose']['enabled']=False;settings['tracking']['enabled']=True;settings['intrusion']['enabled']=True
  settings['display_categories']=base['settings']['display_categories'];settings['debug_all_candidates']=False
  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']
  body={k:v for k,v in spec.items() if k in allowed};body.update(stream_id=sid,source_id=str(ident),camera_id=ident,store=False,active_learning=False,autostart=False,scenarios='off',analysisFpsTarget=10)
  base['camera'].update(stream_id=sid,source_id=ident);base['test_label']=old[sid]['source_id']+'; input='+old[origin]['source_id']
  base['settings'].update(tracking={'enabled':True},intrusion=settings['intrusion'],display_categories=settings['display_categories'])
  base['debug_metrics']=False
  file=(presetroot/('web-cam08.yaml' if i==0 else 'web-cam01.yaml')) if i<2 else presetroot/'medium16'/(sid+'.yaml')
  rows.append(dict(stream_id=sid,source_id=str(ident),camera_id=ident,group=group+1,copy=copies[group],original=i<2,origin=origin,test_label=base['test_label'],preset=str(file)))
  plans.append(dict(sid=sid,old=old[sid],body=body,settings=settings,preset=base,path=str(file)))
 save('migration-plan.json',dict(rows=rows,plans=plans,presets_before={str(f):f.read_text() for f in presetroot.rglob('*.yaml')}))
 for item in plans:
  sid=item['sid'];q=rt.states()[sid];assert q['state']=='stopped' and not q.get('child_pid') and not q.get('cleanup_pending')
  save('delete-intent-'+sid+'.json',dict(run=q.get('run_id'),generation=q.get('run_generation')))
  rt.api.call('DELETE','/api/v1/streams/'+quote(sid,safe=''))
  save('register-intent-'+sid+'.json',item['body']);rt.api.call('POST','/api/v1/streams?start=false',item['body'])
  revision=rt.settings(sid)['revision'];patch=dict(expected_revision=revision,**item['settings'])
  save('settings-intent-'+sid+'.json',patch);rt.api.call('PATCH','/api/v1/streams/'+quote(sid,safe='')+'/settings',patch)
  actual=rt.settings(sid);assert actual['tracking']['enabled'] and actual['intrusion']['enabled'] and not actual['fire']['enabled'] and not actual['pose']['enabled']
  file=Path(item['path']);file.parent.mkdir(parents=True,exist_ok=True);file.write_text(yaml.safe_dump(item['preset'],sort_keys=False,allow_unicode=True));file.chmod(0o600);load_preset(file)
  save('migrated-'+sid+'.json',dict(state=rt.states()[sid],settings=actual,preset_hash=file_hash(file)))
 manifest=Path('/workspace/contextra-runtime-config/test-run/medium16-registry.json');shutil.copy2(manifest,R/'manifest.before')
 manifest.write_text(json.dumps(dict(channels=rows,purpose='Reusable two Web inputs, eight channels each',source_commit=p['head'],instance_candidate=8,max_batch=4,presets=str(presetroot/'medium16')),ensure_ascii=False,indent=2));manifest.chmod(0o600)
 # Replace the old sample01-specific convenience preset with the new numeric registration.
 extra=presetroot/'intrusion-demo.yaml'
 if extra.exists():
  item=next(v for v in plans if v['sid']=='rtsp-test-01');extra.write_text(yaml.safe_dump(item['preset'],sort_keys=False,allow_unicode=True));extra.chmod(0o600)
 cfg=copy.deepcopy(p['config']);cfg.update(expected_image=read('image.json')['id'],model=dict(name='RFDETRMedium',version=4,role='det',instances=8,max_batch=4,queue_delay_us=5000),
  settings_expected={'analysisFpsTarget':10,'tracking':{'enabled':True},'intrusion':{'enabled':True},'fire':{'enabled':False},'pose':{'enabled':False},'display_score':.25,'display_categories':['person']},failure_stop_all=True)
 new=Path('/workspace/contextra-runtime-config/test-run/event-artifact.yaml');new.write_text(yaml.safe_dump(cfg,sort_keys=False));new.chmod(0o600)
 actual=yaml.safe_load(REG.read_text());current={v['stream_id']:v for v in actual['added']}
 for sid,s in old.items():
  if sid not in target:assert current[sid]==s
 save('migrated.json',dict(rows=rows,states=rt.states(),registry_hash=file_hash(REG),config=str(new)))
 print('Migrated sixteen numeric trial IDs; two inputs x8; all stopped',flush=True)
def events(rt):
 code="""import json
from app.event_artifact_contract import ArtifactConfig
from app.event_artifact_store import ArtifactStore
s=ArtifactStore(ArtifactConfig.from_environment())
with s.transaction() as q:
 q.execute('SELECT * FROM contextra_event_artifacts ORDER BY created_at DESC LIMIT 100')
 print(json.dumps([dict(r) for r in q.fetchall()]))
"""
 return json.loads(rt.command(['docker','exec',rt.container,'python3','-c',code],15))
def stop_targets(rt):
 result={}
 for row in read('migrated.json')['rows']:
  sid=row['stream_id'];q=rt.states()[sid]
  try:
   if q.get('child_pid') or q['state'] in ['running','starting']:result[sid]=rt.stop(sid)
   else:result[sid]=q
  except Exception as e:result[sid]=dict(error=type(e).__name__,state=rt.states().get(sid))
 save('failure-stop.json',result)
def trial():
 import urllib.request
 cfg=load_config(Path(read('migrated.json')['config']));rt=Runtime(cfg,R)
 assert not (R/'trial-start.json').exists();save('trial-start.json',dict(at=time.time(),caps=read('plan.json')['caps']))
 rows=read('migrated.json')['rows'][:2];samples=[];failure=None
 try:
  for row in rows:
   q=rt.start(row['stream_id']);save('first-frame-'+row['stream_id']+'.json',q)
   assert q['metadata_path'].get('artifacts') is not None
  time.sleep(5)
  begin=time.monotonic()
  while time.monotonic()-begin<65:
   sample=rt.sample(rows);sample['events']=events(rt);sample['states']=rt.states();samples.append(sample);save('trial-observation.json',samples)
   for sid,q in sample['channels'].items():
    if q['state']!='running' or q['error'] or q['stale'] or q['drop'] or q['processed']>1000:raise RuntimeError('Runtime bound or Frame degradation')
    art=sample['states'][sid]['metadata_path']['artifacts']
    if art['error'] or art.get('unaccepted_events',0):raise RuntimeError('Artifact admission failure')
   if any(e['artifact_state']=='FAILED' or e['delivery_state']=='FAILED' for e in sample['events']):
    raise RuntimeError('Artifact or Web delivery boundary failed')
   time.sleep(10)
  if not samples[-1]['events']:failure='NATURAL_EVENT_NOT_OBSERVED'
  receipts=[]
  for sid in [r['stream_id'] for r in rows]:
   e=next((e for e in samples[-1]['events'] if e['payload']['streamId']==sid and e['artifact_state']=='READY'),None)
   if not e:continue
   for kind in ['clipUrl','thumbnailUrl']:
    body=e.get('body') or {};wire=(body.get('events') or [{}])[0];url=wire.get(kind)
    if not url:
     url=rt.inspect()['Config']['Env'];url=next(v.split('=',1)[1] for v in url if v.startswith('CONTEXTRA_ARTIFACT_PUBLIC_BASE='))+'/'+e['relative_path']+'/'+('clip.mp4' if kind=='clipUrl' else 'thumbnail.jpg')
    with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(url,timeout=5) as res:
     payload=res.read(16*1024**2);receipts.append(dict(event_id=e['event_id'],kind=kind,status=res.status,size=len(payload),sha256=hashlib.sha256(payload).hexdigest(),url=url))
  save('http-get.json',receipts)
 except Exception as exc:
  failure=type(exc).__name__+': '+str(exc)
  save('trial-failure.json',dict(error=failure,states=rt.states(),events=events(rt)))
  stop_targets(rt)
 finally:
  save('trial-final.json',dict(failure=failure,states=rt.states(),events=events(rt),resources=rt.resources()))
 print('Trial finished; failure='+str(failure),flush=True)
def close_failure():
 import urllib.request
 cfg=load_config(Path(read('migrated.json')['config']));rt=Runtime(cfg,R);found=events(rt);save('events-before-stop.json',found)
 receipts=[]
 for camera in [77,78]:
  e=next((e for e in found if e['payload']['cameraId']==camera and e['artifact_state']=='READY'),None)
  if not e:continue
  for name in ['clip.mp4','thumbnail.jpg']:
   url='http://192.168.50.221:35001/artifacts/'+e['relative_path']+'/'+name
   try:
    with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(url,timeout=5) as res:
     payload=res.read(16*1024**2);receipts.append(dict(cameraId=camera,eventId=e['event_id'],file=name,status=res.status,size=len(payload),sha256=hashlib.sha256(payload).hexdigest(),url=url))
   except Exception as ex:receipts.append(dict(cameraId=camera,file=name,error=type(ex).__name__))
 save('http-get.json',receipts)
 save('states-before-container-stop.json',rt.states())
 assert not (R/'container-stop-intent.json').exists();save('container-stop-intent.json',dict(id=rt.inspect()['Id'],at=time.time()))
 cmd(['docker','stop','-t','80',rt.container],95)
 save('container-stopped.json',dict(analytics=rt.inspect(),redis=json.loads(cmd(['docker','inspect','contextra-redis-1']))[0]['State'],postgres=json.loads(cmd(['docker','inspect','contextra-postgresql-1']))[0]['State']))
 print('GET checked; Analytics stopped, no rollback. Redis/PostgreSQL preserved.',flush=True)
def finalize():
 import csv
 cfgpath=Path(read('migrated.json')['config']);cfg=yaml.safe_load(cfgpath.read_text())
 cfg['stages']=[dict(channels=2,seconds=60)]
 source=Path('/data/contextra/model-artifacts/deployments/medium-fp16-4471edc/RFDETRMedium')
 cfg['artifacts']=[dict(name=n,path=str(f),sha256=file_hash(f)) for n,f in [('engine',source/'4/model.plan'),('model-config',source/'config.pbtxt'),('registry',REG)]]
 cfgpath.write_text(yaml.safe_dump(cfg,sort_keys=False));cfgpath.chmod(0o600)
 upcoming=copy.deepcopy(cfg);upcoming['expected_image']=read('final-image.json')['id']
 nextpath=cfgpath.with_name('event-artifact-next.yaml');nextpath.write_text(yaml.safe_dump(upcoming,sort_keys=False));nextpath.chmod(0o600)
 states=read('states-before-container-stop.json');measure=read('trial-observation.json');summary=[]
 for row in read('migrated.json')['rows'][:2]:
  sid=row['stream_id'];end=measure[-1]['states'][sid];first=read('first-frame-'+sid+'.json');span=(end['child_sampled_ns']-first['child_sampled_ns'])/1e9
  q=measure[-1]['channels'][sid];summary.append(dict(cameraId=row['camera_id'],stream_id=sid,reference_startup_seconds=span,
   reference_startup_result_fps=(end['metadata_path']['processed']-first['metadata_path']['processed'])/span if span else None,
   steady_result_fps=None,steady_redis_fps=None,queue=q['queue'],stale=q['stale'],drop=q['drop'],cleanup_completed=states[sid]['last_cleanup']['completed'],child_reaped=states[sid]['last_cleanup']['child_reaped']))
 save('summary.json',dict(channels=summary,trial='STOPPED_WEB_HTTP_404',events=18,artifact_ready=18,web_accepted=0,
  runtime_image=read('image.json'),prepared_image=read('final-image.json'),steady_fps='not measured: one initial sample, early stop',
  no_rollback=True,actual_config=str(cfgpath),next_config=str(nextpath)))
 with (R/'summary.csv').open('w') as f:
  w=csv.DictWriter(f,fieldnames=list(summary[0]));w.writeheader();w.writerows(summary)
 print('Final evidence and private configs aligned; next image remains inactive',flush=True)
if __name__=='__main__':globals()[sys.argv[1]](*sys.argv[2:])