import importlib.util, json, os, pathlib, selectors, subprocess, time
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=a.built;C=a.C;cmd=a.cmd;api=a.api
def monitor_startup():
 p=subprocess.Popen(['docker','logs','--follow',g.NAME],stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
 sel=selectors.DefaultSelector();sel.register(p.stdout,selectors.EVENT_READ)
 size=0;start=time.monotonic();checked=0;ready=False
 try:
  with (R/'startup.private.log').open('xb') as log:
   while time.monotonic()-start<300:
    for key,_ in sel.select(.2):
     data=os.read(key.fileobj.fileno(),65536)
     if not data:sel.unregister(key.fileobj);continue
     assert size+len(data)<2*1024*1024,'LOG_CAP'
     log.write(data);log.flush();size+=len(data)
    if time.monotonic()-checked<1:continue
    checked=time.monotonic();c=g.inspect()
    assert c['Image']==built['image_id'] and c['State']['Status']=='running'
    assert not c['RestartCount'] and not c['State']['OOMKilled']
    assert g.gpu()-plan['gpu_mib']<8192
    mem={l.split(':')[0]:int(l.split()[1]) for l in pathlib.Path('/proc/meminfo').read_text().splitlines()}
    assert mem['MemAvailable']>4*1024**2
    try:
     ready=c['State'].get('Health',{}).get('Status')=='healthy' and all(api(x).get('ready',True) for x in ['/live','/ready','/health'])
    except Exception:ready=False
    if ready:break
  assert ready,'READINESS_TIMEOUT'
  return dict(bytes=size,elapsed_s=time.monotonic()-start)
 finally:
  sel.close()
  if p.poll() is None:p.terminate()
  try:p.wait(5)
  except subprocess.TimeoutExpired:p.kill();p.wait(5)
def main():
 assert not (R/'apply.json').exists()
 state=dict(stage='starting',runtime_windows=0)
 def save():(R/'apply.json').write_text(json.dumps(state,indent=2))
 save()
 try:
  assert json.loads((R/'image-verify.json').read_text())['errors']==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 json.loads((R/'source.json').read_text())['hashes'].items())
  assert g.sha(g.ENV)==plan['env_hash'] and g.sha(a.overlay)==plan['overlay_hash']
  assert g.inspect()['Id']==plan['analytics']['Id']
  r=json.loads(g.sh(['docker','inspect','contextra-redis-1']))[0];assert r['Id']==plan['redis']['Id']
  assert g.sh(['docker','exec',r['Id'],'redis-cli','DBSIZE']).strip()=='0'
  assert all(g.sha(p)==h for p,h in plan['control_hashes'].items())
  assert api('/api/v1/streams')['streams'][0]['state']=='stopped'
  assert not pathlib.Path(plan['unit']).exists()
  cmd(['install','-m','644',str(R/'contextra-metadata-firewall.service'),plan['unit']])
  state['firewall_applied']=True;save()
  cmd(['systemd-analyze','verify',plan['unit']]);cmd(['systemctl','daemon-reload'])
  cmd(['systemctl','enable','--now','contextra-metadata-firewall.service'])
  cmd(['python3',str(C/'firewall.py'),'check'])
  lines=g.ENV.read_text().splitlines(keepends=True)
  indices=[i for i,x in enumerate(lines) if x.split('=',1)[0]=='CONTEXTRA_IMAGE'];assert len(indices)==1
  lines[indices[0]]='CONTEXTRA_IMAGE='+built['tag']+'\n'
  state['env_after']=a.atomic(g.ENV,''.join(lines).encode(),plan['env_hash']);save()
  state['redis_reflected']=True;save()
  cmd(plan['compose']+['up','-d','--no-deps','--force-recreate','--pull','never','redis'],120)
  end=time.monotonic()+60
  while time.monotonic()<end:
   r=json.loads(g.sh(['docker','inspect','contextra-redis-1']))[0]
   assert not r['RestartCount'] and r['State']['Status']=='running'
   if r['State'].get('Health',{}).get('Status')=='healthy':break
   time.sleep(1)
  assert r['State']['Health']['Status']=='healthy'
  assert g.sh(['docker','exec',r['Id'],'redis-cli','--raw','PING']).startswith('NOAUTH')
  state.update(redis_id=r['Id'],redis_auth='PASS');save()
  state['analytics_reflected']=True;save()
  cmd(plan['compose']+['up','--no-start','--no-deps','--force-recreate','--pull','never','analytics'],120)
  c=g.inspect();assert c['State']['Status']=='created' and c['Image']==built['image_id']
  assert c['HostConfig']['RestartPolicy']['Name']=='no'
  cmd(['docker','start',g.NAME],30)
  state['capture']=monitor_startup()
  assert api('/api/v1/streams')['streams'][0]['state']=='stopped'
  assert api('/api/v1/metadata/status')['enabled'] is True
  state.update(stage='ready',analytics_id=g.inspect()['Id'],metadata=api('/api/v1/metadata/status'))
  save();print(json.dumps(dict(stage='ready',redis_auth='PASS',analytics='healthy',stream='stopped',capture=state['capture'])))
 except Exception as error:
  state.update(stage='failed',error=type(error).__name__);save()
  try:
   c=g.inspect();state['failed_state']=c['State'];state['failed_image']=c['Image']
   logs=subprocess.run(['docker','logs','--tail','150',g.NAME],capture_output=True,timeout=10)
   (R/'failure.private.log').write_bytes((logs.stdout+logs.stderr)[-1024*1024:])
  except Exception as e:state['observation_error']=type(e).__name__
  save()
  try:
   a.rollback(state)
   cmd(['docker','update','--restart='+a.restart_arg(plan['analytics']['HostConfig']['RestartPolicy']),g.NAME])
  except Exception as e:
   state['rollback_error']=type(e).__name__
   cmd(['docker','stop','--timeout','20',g.NAME],40)
  save();print(json.dumps({k:state.get(k) for k in ['stage','error','rollback','rollback_error']}))
  raise SystemExit(1)
if __name__=='__main__':main()