import gate as g
import os,json,time,pathlib,urllib.request,urllib.parse,signal,subprocess,stat
os.umask(0o077);R=g.R
gate=json.loads((R/'gate.json').read_text());assert not (R/'runtime.json').exists()
env=g.load_env_file(str(g.ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
spec=g.yaml.safe_load(overlay.read_text())['added'][0]
base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
path='/api/v1/streams/'+urllib.parse.quote(spec['stream_id'],safe='')
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
result={'stage':'prepared','start_attempts':0,'stop_attempts':0,'readiness_budget_s':300,'frame_budget':100,
 'first_frame_budget_s':60,'after_first_s':10,'cleanup_http_budget_s':25,'gpu_increment_cap_mib':8192}
changed=False;start=None
def save():
 p=R/'runtime.partial';p.write_text(json.dumps(result,indent=2));p.replace(R/'runtime.json')
def api(p,method='GET',timeout=5):
 with opener.open(urllib.request.Request(base+p,method=method),timeout=timeout) as f:return json.load(f)
def faildata(v):
 if not isinstance(v,dict):return None
 return {k:v.get(k) for k in ['error_code','failure_stage','role','exception_type'] if k in v}
def state(v):
 d={k:v.get(k) for k in ['state','desired_state','observed_state','cleanup_pending','run_generation','per_task_buffers']}
 d['last_error']=faildata(v.get('last_error'));d['cleanup_error']=faildata(v.get('cleanup_error'))
 d['roles']={k:{x:val.get(x) for x in ['state','source_stage','stage_counts','stage_first_at']} for k,val in v.get('roles',{}).items() if k in ['det','pose','seg']}
 return d
def cas(policy):
 c=g.inspect();assert c['Id']==gate['container_id'] and c['Image']==gate['image']
 assert c['HostConfig']['RestartPolicy']==policy
 assert g.sha(g.ENV)==gate['env_hash'] and g.sha(g.CFG)==gate['config_hash'] and g.sha(overlay)==gate['overlay_hash']
 return c
def stop_container():
 c=g.inspect();assert c['Id']==gate['container_id']
 if c['State']['Status']=='running':
  subprocess.run(['docker','stop','--timeout','20',g.NAME],check=True,capture_output=True,timeout=35)
 result['analytics_safety_stop']=True;save()
def threads():
 code='''import pathlib,json,os,stat
v={'processes':0,'tasks':0,'media':0,'stdin_kind':None}
for p in pathlib.Path('/proc').glob('[0-9]*'):
 try:
  a=(p/'cmdline').read_bytes().split(bytes([0]))
  if len(a)<3 or a[1:3]!=[b'-m',b'app.main']:continue
  v['processes']+=1
  mode=os.stat(p/'fd/0').st_mode
  v['stdin_kind']='pipe' if stat.S_ISFIFO(mode) else ('character_device' if stat.S_ISCHR(mode) else 'other')
  for q in (p/'task').iterdir():
   name=(q/'comm').read_text().lower();v['tasks']+=1
   if any(k in name for k in ['rtp','rtsp','infer','preprocess','appsink','nvvideoconvert']):v['media']+=1
 except (OSError,ProcessLookupError):pass
print(json.dumps(v))'''
 return json.loads(g.sh(['docker','exec',g.NAME,'python3','-c',code]))
def alarm(*args):raise TimeoutError('Overall deadline')
signal.signal(signal.SIGALRM,alarm);signal.alarm(480)
save()
try:
 assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()==g.HEAD
 c=cas(gate['original_restart']);assert c['State']['Status']=='exited'
 assert g.others()==gate['other_containers']
 subprocess.run(['docker','update','--restart=no',g.NAME],check=True,capture_output=True,timeout=15)
 changed=True;result['restart_temporarily_disabled']=True;save()
 c=cas({'Name':'no','MaximumRetryCount':0})
 subprocess.run(['docker','start',g.NAME],check=True,capture_output=True,timeout=30)
 result['stage']='readiness';save();end=time.monotonic()+300
 ready=False
 while time.monotonic()<end:
  c=cas({'Name':'no','MaximumRetryCount':0})
  assert c['State']['Status']=='running' and not c['State']['OOMKilled'] and not c['RestartCount']
  assert g.gpu()-gate['gpu_mib']<8192
  try:
   for endpoint in ['/live','/ready','/health']:api(endpoint)
   ready=c['State'].get('Health',{}).get('Status')=='healthy'
   if ready:break
  except Exception:pass
  time.sleep(2)
 assert ready
 versions=json.loads(g.sh(['docker','exec',g.NAME,'python3','-c',
 'import urllib.request,json;d=json.load(urllib.request.urlopen("http://localhost:8000/v2/models/RFDETRMedium",timeout=5));print(json.dumps(d["versions"]))']))
 assert versions==['2']
 listed=api('/api/v1/streams')['streams'];assert len(listed)==1 and listed[0]['state']=='stopped'
 initial=api(path);assert initial['state']=='stopped' and initial['run_generation']==0 and initial.get('run_id') is None
 result.update(stage='ready',model_versions=versions,threads_before=threads(),gpu_ready_mib=g.gpu());save()
 cas({'Name':'no','MaximumRetryCount':0})
 start=time.monotonic();result.update(stage='window',start_attempts=1);save()
 first=None;success=False
 try:
  api(path+'/start','POST',10)
  while time.monotonic()-start<60:
   v=api(path,timeout=3);result['last_state']=state(v)
   frames=(v.get('per_task_buffers') or {}).get('det',0);result['frames']=frames
   used=g.gpu();result['gpu_peak_mib']=max(result.get('gpu_peak_mib',used),used)
   if used-result['gpu_ready_mib']>8192:result['gpu_cap_exceeded']=True;break
   if v.get('last_error'):break
   if frames and v.get('observed_state')=='running':
    if first is None:first=time.monotonic();result['first_observed_s']=round(first-start,3)
    if frames>=100 or time.monotonic()-first>=10:success=True;break
   save();time.sleep(.5)
 except Exception as e:result['window_exception']=type(e).__name__
 finally:
  result['first_frame_confirmed']=first is not None
  result['stop_attempts']=1;result['stop_at_s']=round(time.monotonic()-start,3);save()
  try:
   response=api(path+'/stop','POST',25);result['stop_response']=state(response)
   result['stop_returned_s']=round(time.monotonic()-start,3)
   post=api(path);result['post_stop']=state(post)
   result['cleanup_verified']=all(post.get(k)=='stopped' for k in ['state','desired_state','observed_state']) and not post.get('cleanup_pending') and not post.get('cleanup_error')
   result['threads_after']=threads()
   result['worker_cleanup_verified']=result['cleanup_verified'] and result['threads_after']['media']<=result['threads_before']['media']
   if not result['worker_cleanup_verified']:stop_container()
  except Exception as e:
   result.update(cleanup_verified=False,cleanup_exception=type(e).__name__);stop_container()
 result['stage']='observed';result['delivery_window_complete']=success
except Exception as e:
 result.update(stage='failed',exception_type=type(e).__name__)
 if changed:stop_container()
finally:
 signal.alarm(0)
 if changed:
  try:
   c=cas({'Name':'no','MaximumRetryCount':0})
   if c['State']['Status']=='running' and not result.get('worker_cleanup_verified'):stop_container()
   subprocess.run(['docker','update','--restart=on-failure:3',g.NAME],check=True,capture_output=True,timeout=15)
   result['restart_restored']=g.inspect()['HostConfig']['RestartPolicy']==gate['original_restart']
  except Exception as e:result['restart_restore_error']=type(e).__name__
 d=pathlib.Path(gate['diagnostics_dir']);rows=[]
 for p in d.glob('*.jsonl'):
  if p.name in gate['diagnostics_before']:continue
  for line in p.read_text().splitlines():
   try:
    v=json.loads(line);rows.append({k:v.get(k) for k in ['stage','phase','thread_role','monotonic_ns','diagnostic_error']})
   except ValueError:pass
 result['new_markers']=sorted(rows,key=lambda v:v['monotonic_ns'])[:100]
 result['historical_diagnostics_unchanged']=all((d/k).exists() and g.sha(d/k)==h for k,h in gate['diagnostics_before'].items())
 result['diagnostic_override_retained']=True
 result['other_containers_unchanged']=g.others()==gate['other_containers']
 result['env_config_overlay_unchanged']=g.sha(g.ENV)==gate['env_hash'] and g.sha(g.CFG)==gate['config_hash'] and g.sha(overlay)==gate['overlay_hash']
 c=g.inspect();result['analytics']={k:c['State'].get(k) for k in ['Status','OOMKilled','ExitCode']}
 result['analytics']['restart']=c['RestartCount'];result['container_id_unchanged']=c['Id']==gate['container_id']
 result['gpu_final_mib']=g.gpu();save()
 print(json.dumps({k:v for k,v in result.items() if k!='new_markers'},indent=2))