import json,os,pathlib,signal,subprocess,tempfile,time,urllib.parse,urllib.request
from prepare import g,R,HEAD,source_cas
from tools.validate_deployment import validate_deployment_inputs
os.umask(0o077)
gate=json.loads((R/'gate.json').read_text());build=json.loads((R/'build.json').read_text())
assert not (R/'runtime.json').exists()
assert json.loads((R/'image-verify.json').read_text())['status']=='PASS'
source_cas()
env=g.load_env_file(str(g.ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
s=g.yaml.safe_load(overlay.read_text())['added'][0]
path='/api/v1/streams/'+urllib.parse.quote(s['stream_id'],safe='')
base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
result=dict(stage='prepared',calls=[],restart_mode='explicit_restart_same_config',snapshots=[],
            readiness_budget_s=300,first_frame_budget_s=60,restart_budget_s=125,final_cleanup_budget_s=70,
            helper_budget_s=720,frame_cap=1000,gpu_increment_cap_mib=8192,host_ram_floor_gib=4)
cid=None;changed=False;env_after=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=3):
 with opener.open(urllib.request.Request(base+p,method=method),timeout=timeout) as f:return f.status,json.load(f)
def command(action):
 row={'action':action,'requested_ns':time.monotonic_ns()};result['calls'].append(row);save()
 code,value=api(path+'/'+action,'POST',10)
 row.update(http=code,returned_ns=time.monotonic_ns(),state=value.get('state'));save()
 return value
def current():
 c=g.inspect()
 if cid is not None:assert c['Id']==cid and c['Image']==build['image_id']
 return c
def common_cas():
 assert g.sha(g.CFG)==gate['config_hash'] and g.sha(overlay)==gate['overlay_hash']
 assert g.sha(g.CFG.parent/'2/model.plan')==gate['plan_hash']
def resources():
 used=g.gpu()
 mem={l.split(':')[0]:int(l.split()[1]) for l in pathlib.Path('/proc/meminfo').read_text().splitlines()}
 assert used-gate['gpu_mib']<8192 and mem['MemAvailable']>4*1024**2,'RESOURCE_CAP'
 result['gpu_peak_mib']=max(result.get('gpu_peak_mib',used),used)
def snapshot():
 _,v=api(path)
 row={k:v.get(k) for k in ['state','desired_state','observed_state','run_generation','cleanup_pending','restart_pending','last_cleanup','per_task_buffers','runtime_observation']}
 for name in ['last_error','cleanup_error']:
  x=v.get(name);row[name]={k:x.get(k) for k in ['error_code','failure_stage','role']} if isinstance(x,dict) else None
 row['monotonic_ns']=time.monotonic_ns()
 for observation in v.get('runtime_observation') or []:
  assert observation['generation']==v['run_generation']
  assert max((x['count'] for x in observation['counters'].values()),default=0)<=1000,'FRAME_CAP'
 resources();result['snapshots'].append(row);save()
 return v
def wait_running(generation,budget):
 end=time.monotonic()+budget
 while time.monotonic()<end:
  v=snapshot()
  assert not v.get('last_error') and not v.get('cleanup_error'),'PIPELINE_ERROR'
  if v['run_generation']==generation and v['observed_state']=='running' and (v.get('per_task_buffers') or {}).get('det',0)>0:
   return v
  time.sleep(.5)
 raise TimeoutError('FIRST_FRAME_OR_RESTART_DEADLINE')
def safety_stop():
 c=current()
 if c['State']['Status']=='running':subprocess.run(['docker','stop','--timeout','20',g.NAME],capture_output=True,check=True,timeout=35)
 result['analytics_safety_stop']=True;save()
def alarm(*unused):raise TimeoutError('HELPER_DEADLINE')
signal.signal(signal.SIGALRM,alarm);signal.alarm(720);save()
try:
 common_cas();assert g.sha(g.ENV)==gate['env_hash']
 c=g.inspect();assert c['Id']==gate['container_id'] and c['Image']==gate['old_image'] and c['State']['Status']=='exited'
 assert c['HostConfig']['RestartPolicy']==gate['original_restart'] and g.others()==gate['other_containers']
 from model_tooling.deploy import deployment_plan
 from model_tooling.medium_acceptance import POLICY
 plan=deployment_plan(pathlib.Path('/data/contextra/model-artifacts/builds/medium-193-20260909-v4'),R/'unapplied-target',1,source_acceptance=POLICY)
 assert plan['source_acceptance']['status']=='LIMITED_USER_VISUAL_ACCEPTANCE'
 diag='/run/contextra-start-diagnostics'
 override={'services':{'analytics':{'restart':'no','environment':{'CONTEXTRA_START_DIAGNOSTICS_DIR':diag},
           'volumes':[{'type':'bind','source':str(R/'diagnostics'),'target':diag,'bind':{'create_host_path':False}}]}}}
 (R/'diagnostics-compose.yaml').write_text(g.yaml.safe_dump(override))
 compose=['docker','compose','--env-file',str(g.ENV),'-f',str(g.REPO/'docker-compose.yml'),'-f',str(R/'diagnostics-compose.yaml')]
 lines=g.ENV.read_text().splitlines(keepends=True);indices=[i for i,x in enumerate(lines) if x.split('=',1)[0].strip()=='CONTEXTRA_IMAGE'];assert len(indices)==1
 lines[indices[0]]='CONTEXTRA_IMAGE='+build['tag']+'\n';assert g.sha(g.ENV)==gate['env_hash']
 fd,name=tempfile.mkstemp(dir=g.ENV.parent)
 with os.fdopen(fd,'w') as f:os.fchmod(f.fileno(),0o600);f.writelines(lines);f.flush();os.fsync(f.fileno())
 os.replace(name,g.ENV);changed=True;env_after=g.sha(g.ENV);result.update(stage='env_applied',env_after_hash=env_after);save()
 validate_deployment_inputs(str(g.ENV))
 subprocess.run(compose+['config','--quiet'],capture_output=True,check=True,timeout=30)
 render=json.loads(subprocess.check_output(compose+['config','--format','json'],text=True,timeout=30))['services']['analytics']
 assert render['image']==build['tag'] and render['restart']=='no'
 common_cas();assert g.sha(g.ENV)==env_after
 subprocess.run(compose+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],capture_output=True,check=True,timeout=120)
 cid=g.inspect()['Id'];result['stage']='readiness';save();end=time.monotonic()+300;ready=False
 while time.monotonic()<end:
  c=current();resources()
  assert c['State']['Status']=='running' and not c['RestartCount'] and not c['State']['OOMKilled']
  assert c['HostConfig']['RestartPolicy']=={'Name':'no','MaximumRetryCount':0}
  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')[1]['streams'];assert len(listed)==1 and listed[0]['state']=='stopped'
 initial=api(path)[1];assert initial['run_generation']==0 and initial.get('run_id') is None
 result.update(stage='live',readiness='PASS',model_versions=versions);save()
 try:
  command('start');first=wait_running(1,60);first_id=first['run_id']
  result['first_frame_ns']=time.monotonic_ns();save();time.sleep(3)
  snapshot();command('restart');second=wait_running(2,125)
  assert second['run_id']!=first_id
  cleanup=second['last_cleanup'];assert cleanup['generation']==1 and cleanup['completed'] is True
  observations=second.get('runtime_observation') or [];assert len(observations)==1
  assert observations[0]['events']['start_enter']>=cleanup['finished_monotonic_ns']
  result.update(restart_verified=True,run_id_changed=True,no_overlap_by_cleanup_order=True,second_frame_ns=time.monotonic_ns(),restart_cleanup=cleanup);save()
  time.sleep(3);snapshot()
 except Exception as error:result['window_error']=type(error).__name__;save()
 finally:
  command('stop');end=time.monotonic()+70
  while time.monotonic()<end:
   v=snapshot()
   if v.get('cleanup_error') or v.get('last_error'):break
   if v['state']=='stopped' and not v['cleanup_pending']:
    assert not v.get('runtime_observation') and not v.get('restart_pending')
    result['final_cleanup_verified']=True;result['final_cleanup']=v['last_cleanup'];break
   time.sleep(.5)
  if not result.get('final_cleanup_verified') or not result.get('restart_verified'):safety_stop()
  else:
   for endpoint in ['/live','/ready','/health']:api(endpoint)
   assert current()['State'].get('Health',{}).get('Status')=='healthy'
 result['stage']='completed'
except Exception as error:
 result.update(stage='failed',exception_type=type(error).__name__);save()
 if changed:safety_stop()
finally:
 signal.alarm(0)
 if changed:
  try:
   c=current();assert c['HostConfig']['RestartPolicy']=={'Name':'no','MaximumRetryCount':0}
   subprocess.run(['docker','update','--restart=on-failure:3',g.NAME],capture_output=True,check=True,timeout=15)
   result['restart_restored']=current()['HostConfig']['RestartPolicy']==gate['original_restart']
  except Exception as error:result['restore_error']=type(error).__name__
 result['other_containers_unchanged']=g.others()==gate['other_containers']
 result['model_config_overlay_unchanged']=g.sha(g.CFG)==gate['config_hash'] and g.sha(overlay)==gate['overlay_hash'] and g.sha(g.CFG.parent/'2/model.plan')==gate['plan_hash']
 c=current();result['analytics']={k:c['State'].get(k) for k in ['Status','ExitCode','OOMKilled']};result['analytics']['restart_count']=c['RestartCount']
 result['gpu_final_mib']=g.gpu();result['env_cas_preserved']=env_after is None or g.sha(g.ENV)==env_after
 result['diagnostic_override_retained']=changed;save()
 print(json.dumps({k:result.get(k) for k in ['stage','calls','restart_verified','final_cleanup_verified','analytics','restart_restored','other_containers_unchanged','model_config_overlay_unchanged','window_error','exception_type']},indent=2))