import importlib.util,json,os,pathlib,selectors,signal,subprocess,threading,time,urllib.parse,urllib.request
from build import R,g,HEAD
os.umask(0o077)
s=importlib.util.spec_from_file_location('prior_apply','/tmp/codex/ltb1021/2026-09-11_contextra-metadata-auth_r01/apply.py')
a=importlib.util.module_from_spec(s);s.loader.exec_module(a)
plan=a.plan;built=json.loads((R/'image-r02/build.json').read_text());C=a.C;cmd=a.cmd
env=g.load_env_file(str(g.ENV));base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
spec=g.yaml.safe_load(a.overlay.read_text())['added'][0]
path='/api/v1/streams/'+urllib.parse.quote(spec['stream_id'],safe='')
key=g.hashlib.sha256(spec['source_id'].encode()).hexdigest()
stream='contextra:metadata:v1:camera:'+key+':frames'
state=dict(stage='prepared',calls=[],observations=[],runtime_windows=0)
def save():
 t=R/'run.partial';t.write_text(json.dumps(state,indent=2));t.replace(R/'run.json')
def api(p,method='GET',body=None,timeout=3):
 req=urllib.request.Request(base+p,data=None if body is None else json.dumps(body).encode(),method=method,headers={'Content-Type':'application/json'})
 with opener.open(req,timeout=timeout) as f:return f.status,json.load(f)
def resource():
 c=g.inspect();assert c['Image']==built['image_id'] and c['State']['Status']=='running' and not c['RestartCount']
 assert not c['State']['OOMKilled'] and g.gpu()-plan['gpu_mib']<8192
 m={l.split(':')[0]:int(l.split()[1]) for l in pathlib.Path('/proc/meminfo').read_text().splitlines()}
 assert m['MemAvailable']>4*1024**2
def sample():
 _,v=api(path)
 row=dict(monotonic_ns=time.monotonic_ns(),state=v['state'],generation=v.get('run_generation'),
  cleanup_pending=v.get('cleanup_pending'),metadata_path=v.get('metadata_path'),last_cleanup=v.get('last_cleanup'),
  runtime_observation=v.get('runtime_observation',[]),roles={},publisher=api('/api/v1/metadata/status')[1])
 for k in ('last_error','cleanup_error'):
  e=v.get(k);row[k]={f:e.get(f) for f in ('error_code','failure_stage','role')} if isinstance(e,dict) else None
 for role,r in v.get('roles',{}).items():
  row['roles'][role]={k:r.get(k) for k in ('state','initialized','source_stage','stage_counts')}
 for ob in row['runtime_observation']:
  assert set(ob)<=set(['token','diagnostic_error','events','counters','stop_matches_started','role','generation','worker_alive','sampled_ns'])
  assert max((x['count'] for x in ob['counters'].values()),default=0)<=1000,'FRAME_CAP'
 state['observations'].append(row);save();resource();return v
def command(action):
 row=dict(action=action,requested_ns=time.monotonic_ns());state['calls'].append(row);save()
 code,v=api(path+'/'+action,'POST',timeout=10)
 row.update(http=code,returned_ns=time.monotonic_ns(),state=v.get('observed_state'));save()
class LogCapture:
 def __init__(self,args,target,seconds):
  self.p=subprocess.Popen(args,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
  self.target=target;self.seconds=seconds;self.bytes=0;self.error=None;self.done=threading.Event()
  self.thread=threading.Thread(target=self.run,daemon=True);self.thread.start()
 def run(self):
  sel=selectors.DefaultSelector();sel.register(self.p.stdout,selectors.EVENT_READ);started=time.monotonic()
  try:
   with self.target.open('xb') as log:
    while not self.done.is_set() and time.monotonic()-started<self.seconds:
     for key,_ in sel.select(.1):
      data=os.read(key.fileobj.fileno(),65536)
      if not data:sel.unregister(key.fileobj);continue
      assert self.bytes+len(data)<=2*1024*1024,'LOG_CAP'
      log.write(data);log.flush();self.bytes+=len(data)
     if self.p.poll() is not None and not sel.get_map():return
  except Exception as e:self.error=type(e).__name__
  finally:sel.close()
 def close(self):
  self.done.set()
  if self.p.poll() is None:self.p.terminate()
  try:self.p.wait(5)
  except subprocess.TimeoutExpired:self.p.kill();self.p.wait(5)
  self.thread.join(5);assert not self.thread.is_alive()
def redis_query(role,*args):
 password=json.loads((C/'secrets.json').read_text())[role]
 access=json.loads((C/'access.json').read_text())
 p=subprocess.run(['redis-cli','--json','-h',access['host'],'-p',str(access['port']),
  '--user','contextra_'+role,*args],env=dict(os.environ,REDISCLI_AUTH=password),capture_output=True,text=True,timeout=3)
 assert p.returncode==0,'REDIS_OBSERVATION_FAILED'
 return json.loads(p.stdout)

def read_frames():
 entries=redis_query('reader','XRANGE',stream,'-','+','COUNT','30') or []
 values={}
 for _,fields in entries:
  if isinstance(fields,list):fields=dict(zip(fields[::2],fields[1::2]))
  v=json.loads(fields['json']);assert v['runId']==state['run_id']
  values[v['frameSeq']]=v
 state['redis_frames_seen']=state.get('redis_frames_seen',0)+len(values)
 for f in (R/'visual').glob('capture-*/frame-*.json'):
  v=json.loads(f.read_text());other=values.get(v['frameSeq'])
  if other is not None:
   assert v==other,'FRAME_MISMATCH'
   dst=R/'samples'/f.name
   if not dst.exists():dst.write_text(json.dumps(other))
 save()
def alarm(*unused):raise TimeoutError('HELPER_CAP')

def main():
 assert not (R/'run.json').exists()
 source=json.loads((R/'image-r02/source.json').read_text())
 verified=json.loads((R/'image-r02-verify.json').read_text())
 assert verified['errors']==verified['failures']==verified['sdk']==0
 assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()==HEAD
 assert all(g.sha(g.REPO/p)==h for p,h in source['hashes'].items())
 active=C/'compose.yaml';flag=R/'visual/capture.enabled'
 log=None;started=False;overlay_after=None;active_after=None
 state['calls']=[];state['observations']=[];save()
 signal.signal(signal.SIGALRM,alarm);signal.alarm(600)
 try:
  assert g.sha(active)==plan['active_hash'] and g.sha(g.ENV)==plan['env_hash']
  assert g.sha(a.overlay)==plan['overlay_hash'] and g.inspect()['Id']==plan['analytics']['Id']
  assert g.sha(g.CFG)==plan['config_hash'] and g.sha(g.CFG.parent/'2/model.plan')==plan['plan_hash']
  assert api(path)[1]['state']=='stopped'
  assert redis_query('reader','PING')=='PONG'
  cfg=g.yaml.safe_load(active.read_text());cfg['services']['analytics']['image']=built['image_id']
  cfg['services']['analytics']['environment']['CONTEXTRA_TEST_FRAME_ANCHOR_FILE']=''
  active_after=a.atomic(active,g.yaml.safe_dump(cfg).encode(),plan['active_hash']);state['active_after']=active_after;save()
  lines=g.ENV.read_text().splitlines(keepends=True)
  ix=[i for i,l in enumerate(lines) if l.split('=',1)[0]=='CONTEXTRA_IMAGE'];assert len(ix)==1
  lines[ix[0]]='CONTEXTRA_IMAGE='+built['tag']+'\n'
  state['env_after']=a.atomic(g.ENV,''.join(lines).encode(),plan['env_hash']);save()
  diagnostic={'services':{'analytics':{'restart':'no','environment':{
   'CONTEXTRA_START_DIAGNOSTICS_DIR':'/run/contextra-boundary','CONTEXTRA_METADATA_VISUAL_DIR':'/run/contextra-visual',
   'CONTEXTRA_TEST_FRAME_ANCHOR_FILE':'/run/contextra-test-anchor.json'},'volumes':[
   {'type':'bind','source':str(R/'diagnostics'),'target':'/run/contextra-boundary','bind':{'create_host_path':False}},
   {'type':'bind','source':str(R/'visual'),'target':'/run/contextra-visual','bind':{'create_host_path':False}},
   {'type':'bind','source':str(R/'anchor.json'),'target':'/run/contextra-test-anchor.json','read_only':True}]}}}
  (R/'diagnostic-compose.yaml').write_text(g.yaml.safe_dump(diagnostic))
  compose=['docker','compose','--env-file',str(g.ENV),'-f',str(g.REPO/'docker-compose.yml'),'-f',str(active),'-f',str(R/'diagnostic-compose.yaml')]
  state['compose']=compose;save()
  cmd(compose+['config','--quiet'])
  state['analytics_reflected']=True;save()
  cmd(compose+['up','--no-start','--no-deps','--force-recreate','--pull','never','analytics'],120)
  assert g.inspect()['Image']==built['image_id']
  cmd(['docker','start',g.NAME]);log=LogCapture(['docker','logs','--follow',g.NAME],R/'runtime.private.log',480)
  end=time.monotonic()+300;ready=False
  while time.monotonic()<end:
   resource();assert log.error is None
   try:
    ready=g.inspect()['State'].get('Health',{}).get('Status')=='healthy' and all(api(p)[1].get('ready',True) for p in ('/live','/ready','/health'))
    if ready:break
   except Exception:pass
   time.sleep(2)
  assert ready,'READINESS_TIMEOUT';state['readiness']='PASS';save()
  versions=json.loads(cmd(['docker','exec',g.NAME,'python3','-c','import json,urllib.request;print(json.dumps(json.load(urllib.request.urlopen("http://localhost:8000/v2/models/RFDETRMedium",timeout=3))["versions"]))']).stdout)
  assert versions==['2'];assert sample()['state']=='stopped'
  _,settings=api(path+'/settings')
  _,settings=api(path+'/settings','PATCH',dict(expected_revision=settings['configured']['revision'],display_score=.1))
  assert settings['configured']['display_score']==.1
  overlay_after=g.sha(a.overlay);state['overlay_after']=overlay_after;save()
  (R/'samples').mkdir(mode=0o700);flag.write_text('enabled\n')
  started=True;state['runtime_windows']=1;command('start');end=time.monotonic()+60;first=None
  while time.monotonic()<end:
   v=sample();assert log.error is None
   assert not v.get('last_error') and not v.get('cleanup_error'),'PIPELINE_ERROR'
   if v['state']=='running' and (v.get('metadata_path') or {}).get('processed',0)>0:
    first=v;state['run_id']=v['run_id'];state['first_result_ns']=time.monotonic_ns();break
   time.sleep(.5)
  if first is None:state['window_result']='FIRST_RESULT_TIMEOUT'
  else:
   end=time.monotonic()+8
   while time.monotonic()<end:
    read_frames();sample();assert log.error is None;time.sleep(.5)
   values=[json.loads(f.read_text()) for f in (R/'samples').glob('*.json')]
   expected=json.loads((R/'anchor.json').read_text())
   assert 0<len(values)<=3
   for value in values:
    t=value['frameTime'];assert t['startWallTimeMs']==expected['startWallTimeMs'] and t['rtpClockRate']==expected['rtpClockRate']
    assert t['timeSource']=='SYNTHETIC_TEST_ANCHOR' and t['syncStatus']=='UNAVAILABLE'
    assert t['rtpTimestamp'] is None and t['detectedAtWallMs'] is None
    assert value['ingestTimeWallMs']!=t['startWallTimeMs']
   state.update(window_result='PASS',anchor_transport='PASS',redis_visual_matches=len(values),visual_pairs=len(list((R/'visual').glob('capture-*/*.jpg'))))
  save()
 except Exception as e:
  state['error']=type(e).__name__;state['error_code']=str(e) if str(e) in ('FRAME_CAP','READINESS_TIMEOUT','PIPELINE_ERROR','REDIS_OBSERVATION_FAILED','FRAME_MISMATCH','HELPER_CAP') else 'BOUNDED_STEP_FAILED';save()
 finally:
  signal.alarm(0)
  if started:
   try:
    command('stop');end=time.monotonic()+70
    while time.monotonic()<end:
     v=sample()
     if v.get('cleanup_error') or v.get('last_error'):break
     if v['state']=='stopped' and not v.get('cleanup_pending'):
      state['cleanup_verified']=bool((v.get('last_cleanup') or {}).get('completed'));break
     time.sleep(.5)
   except Exception as e:state['stop_error']=type(e).__name__
  if flag.exists():flag.unlink()
  success=state.get('window_result')=='PASS' and state.get('cleanup_verified') and not state.get('error') and not state.get('stop_error')
  try:
   if not success:
    if state.get('analytics_reflected'):state['failed_state']=g.inspect()['State'];save();cmd(['docker','stop','--timeout','20',g.NAME],40)
    if overlay_after:a.atomic(a.overlay,(R/'overlay.before').read_bytes(),overlay_after)
    if active_after:a.atomic(active,(R/'active-compose.before').read_bytes(),active_after)
    if state.get('env_after'):a.atomic(g.ENV,(R/'env.before').read_bytes(),state['env_after'])
    if state.get('analytics_reflected'):
     args=['docker','compose','--env-file',str(g.ENV)]
     for p,h in plan['old_files'].items():
      assert g.sha(p)==h
      args+=['-f',p]
     cmd(args+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],120)
    state['rollback']='applied'
   cmd(['docker','update','--restart='+a.restart_arg(plan['analytics']['HostConfig']['RestartPolicy']),g.NAME])
   if log:log.close();state['log_bytes']=log.bytes;state['log_error']=log.error
   end=time.monotonic()+100
   while time.monotonic()<end:
    c=g.inspect()
    try:
     if c['State'].get('Health',{}).get('Status')=='healthy' and api('/ready')[1]['ready']:break
    except Exception:pass
    time.sleep(2)
   assert c['State']['Health']['Status']=='healthy' and api(path)[1]['state']=='stopped'
   assert g.others()==plan['other_containers'] and g.sha(g.CFG)==plan['config_hash']
   assert g.sha(g.CFG.parent/'2/model.plan')==plan['plan_hash']
   state['final']=dict(analytics='healthy',stream='stopped',image=c['Image'],success=success,
    metadata_auth_preserved=True,model_non_targets_unchanged=True,restart_restored=True,
    display_score=api(path+'/settings')[1]['configured']['display_score'],capture_off=True,
    test_anchor='RETAIN current container only' if success else 'restored previous environment')
  except Exception as e:
   state['restore_error']=type(e).__name__
   try:cmd(['docker','stop','--timeout','20',g.NAME],40);state['safety_stopped']=True
   except Exception:state['safety_stop_failed']=True
  finally:
   if log:log.close()
  state['stage']='completed';save();print(json.dumps({k:state.get(k) for k in ('stage','window_result','anchor_transport','redis_visual_matches','cleanup_verified','error','error_code','final','restore_error')}))

if __name__=='__main__':main()