import hashlib,json,os,pathlib,signal,subprocess,tempfile,time,urllib.parse,urllib.request,uuid
from gate import g,R as BASE_R
from tools.validate_deployment import validate_deployment_inputs
os.umask(0o077);R=BASE_R/'build-r02'
gate=json.loads((R/'gate.json').read_text());build=json.loads((R/'build.json').read_text())
source=json.loads((R/'source-manifest.json').read_text())
assert not (R/'runtime.json').exists()
assert json.loads((R/'image-verify.json').read_text())['status']=='PASS'
assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()==source['base']
assert all(g.sha(g.REPO/p)==h for p,h in source['edited'].items())
env=g.load_env_file(str(g.ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
document=g.yaml.safe_load(overlay.read_text());s=document['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({}))
identity=overlay.parent/'installation_identity.json'
result=dict(stage='prepared',calls=[],observations=[],readiness_budget_s=300,first_frame_budget_s=60,
            cleanup_budget_s=70,helper_budget_s=600,frame_cap=1000,gpu_increment_cap_mib=8192,
            sample_cap=3,host_ram_floor_gib=4)
reflection=False;env_after=None;overlay_after=None;cid=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=dict(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()
def atomic(path,content,expected):
 assert g.sha(path)==expected
 fd,name=tempfile.mkstemp(dir=path.parent)
 with os.fdopen(fd,'w') as f:
  os.fchmod(f.fileno(),0o600);f.write(content);f.flush();os.fsync(f.fileno())
 assert g.sha(path)==expected
 os.replace(name,path)
 return g.sha(path)
def resources():
 mem={l.split(':')[0]:int(l.split()[1]) for l in pathlib.Path('/proc/meminfo').read_text().splitlines()}
 used=g.gpu();assert mem['MemAvailable']>4*1024**2 and used-gate['gpu_mib']<8192,'RESOURCE_CAP'
 result['gpu_peak_mib']=max(result.get('gpu_peak_mib',used),used)
def status():
 code,v=api(path)
 row={k:v.get(k) for k in ['state','run_generation','cleanup_pending','restart_pending','last_cleanup',
                            'per_task_buffers','metadata_path']}
 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
 observations=v.get('runtime_observation') or []
 for ob in observations:
  assert max((x['count'] for x in ob['counters'].values()),default=0)<=1000,'FRAME_CAP'
 row['monotonic_ns']=time.monotonic_ns();resources();result['observations'].append(row);save();return v
def safety_stop():
 c=g.inspect()
 assert c['Image'] in [gate['old_image'],build['image_id']]
 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(600);save()
try:
 assert g.sha(g.ENV)==gate['env_hash'] and g.sha(overlay)==gate['overlay_hash']
 assert g.sha(g.CFG)==gate['config_hash'] and g.sha(g.CFG.parent/'2/model.plan')==gate['plan_hash']
 c=g.inspect();assert c['Id']==gate['container_id'] and c['Image']==gate['old_image']
 assert c['State']['Status']=='running' and c['HostConfig']['RestartPolicy']==gate['original_restart']
 assert g.others()==gate['other_containers']
 assert api(path)[1]['state']=='stopped'
 assert identity.exists()==gate['identity_exists']
 if identity.exists():
  assert g.sha(identity)==gate['identity_hash'];ident=json.loads(identity.read_text())
 else:
  ident=dict(schemaVersion=1,server_id=s.get('server_id') or str(uuid.uuid4()),sources={})
 mapping=ident['sources'].get(s['stream_id'])
 if mapping is None:
  mapping=dict(source_id=s.get('source_id') or str(uuid.uuid4()),
               reference_kind='EXISTING_CONFIG' if s.get('source_id') else 'AI_INTERNAL',external_camera_id=None)
  ident['sources'][s['stream_id']]=mapping
 if not identity.exists():
  fd=os.open(identity,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
  with os.fdopen(fd,'w') as f:json.dump(ident,f,indent=2);f.flush();os.fsync(f.fileno())
 else:atomic(identity,json.dumps(ident,indent=2),gate['identity_hash'])
 result['identity_registry_hash']=g.sha(identity)
 result['identity_reference_kind']=mapping['reference_kind']
 result['identity_before_fields']={key:s.get(key) for key in ['server_id','source_id']}
 result['identity_before_keys']=[key for key in ['server_id','source_id'] if key in s]
 s['server_id']=ident['server_id'];s['source_id']=mapping['source_id']
 samples=R/'samples';samples.mkdir(mode=0o700)
 enabled=samples/'capture.enabled';enabled.write_text('enabled\n')
 (R/'diagnostics').mkdir(mode=0o700)
 override={'services':{'analytics':{'restart':'no','environment':{
    'CONTEXTRA_METADATA_SAMPLES_DIR':'/run/contextra-metadata-samples',
    'CONTEXTRA_START_DIAGNOSTICS_DIR':'/run/contextra-start-diagnostics'},'volumes':[
    {'type':'bind','source':str(samples),'target':'/run/contextra-metadata-samples','bind':{'create_host_path':False}},
    {'type':'bind','source':str(R/'diagnostics'),'target':'/run/contextra-start-diagnostics','bind':{'create_host_path':False}}]}}}
 (R/'runtime-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/'runtime-compose.yaml')]
 from model_tooling.deploy import deployment_plan
 from model_tooling.medium_acceptance import POLICY
 acceptance=deployment_plan(pathlib.Path('/data/contextra/model-artifacts/builds/medium-193-20260909-v4'),R/'unapplied-target',1,source_acceptance=POLICY)
 assert acceptance['source_acceptance']['status']=='LIMITED_USER_VISUAL_ACCEPTANCE'
 overlay_after=atomic(overlay,g.yaml.safe_dump(document,allow_unicode=True,sort_keys=False),gate['overlay_hash'])
 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'
 env_after=atomic(g.ENV,''.join(lines),gate['env_hash'])
 result.update(stage='config_applied',env_after_hash=env_after,overlay_after_hash=overlay_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'
 assert g.sha(g.ENV)==env_after and g.sha(overlay)==overlay_after
 reflection=True;result['stage']='reflecting';save()
 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=g.inspect();resources()
  assert c['Id']==cid and c['Image']==build['image_id'] and c['State']['Status']=='running'
  assert c['RestartCount']==0 and not c['State']['OOMKilled']
  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'] and status()['state']=='stopped'
 result.update(stage='live',readiness='PASS',model_versions=versions);save()
 try:
  command('start');end=time.monotonic()+60;running=None
  while time.monotonic()<end:
   v=status()
   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:
    running=v;result['first_result_ns']=time.monotonic_ns();break
   time.sleep(.5)
  assert running is not None,'FIRST_RESULT_TIMEOUT'
  end=time.monotonic()+10
  while time.monotonic()<end and len(list(samples.glob('sample-*.json')))<3:
   status();time.sleep(.5)
  values=[json.loads(p.read_text()) for p in sorted(samples.glob('sample-*.json'))]
  assert 1<=len(values)<=3 and all(v['runId']==running['run_id'] for v in values)
  assert all(v['serverId']==ident['server_id'] and v['sourceId']==mapping['source_id'] for v in values)
  assert all(v['frameWidth']>0 and v['frameHeight']>0 and v['normalized'] is True for v in values)
  assert all(v['frameTime']['rtpTimestamp'] is None and v['frameTime']['detectedAtWallMs'] is None for v in values)
  seqs=[v['frameSeq'] for v in values];assert seqs==sorted(set(seqs))
  visible=[];tracks=[]
  for v in values:
   items=v['objects'];assert len(items)<=300
   assert all(0<=o['classId']<80 and 0<=o['confidence']<=1 and not o['eventIds'] for o in items)
   assert all(len(o['bbox'])==4 and 0<=o['bbox'][0]<o['bbox'][2]<=1 and 0<=o['bbox'][1]<o['bbox'][3]<=1 for o in items)
   shown=[o for o in items if o['visible']];visible.append(len(shown))
   tracks.append({o['trackId'] for o in shown if o['trackId'] is not None})
  result.update(sample_count=len(values),sample_identity_match=True,source_dimensions=[values[0]['frameWidth'],values[0]['frameHeight']],
    sample_visible_counts=visible,sample_track_counts=[len(t) for t in tracks],
    shared_visible_track_observed=bool(set.intersection(*tracks)) if tracks else False,
    valid_bbox_observed=any(visible),metadata_verified=True)
  save()
 except Exception as error:result['window_error']=type(error).__name__;save()
 finally:
  command('stop');end=time.monotonic()+70
  while time.monotonic()<end:
   v=status()
   if v.get('last_error') or v.get('cleanup_error'):break
   if v['state']=='stopped' and not v['cleanup_pending']:
    assert not v.get('runtime_observation') and not v.get('restart_pending')
    result['cleanup_verified']=True;result['cleanup']=v['last_cleanup'];break
   time.sleep(.5)
  if not result.get('cleanup_verified') or result.get('window_error'):safety_stop()
 result['stage']='completed'
except Exception as error:
 result.update(stage='failed',exception_type=type(error).__name__);save()
 if reflection:safety_stop()
finally:
 signal.alarm(0)
 enabled=R/'samples/capture.enabled'
 if enabled.exists():
  assert enabled.read_text()=='enabled\n';enabled.unlink();result['sample_capture_disabled']=True
 if reflection:
  try:
   c=g.inspect();assert c['Image']==build['image_id']
   subprocess.run(['docker','update','--restart=on-failure:3',g.NAME],capture_output=True,check=True,timeout=15)
   result['restart_restored']=g.inspect()['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_unchanged']=g.sha(g.CFG)==gate['config_hash'] and g.sha(g.CFG.parent/'2/model.plan')==gate['plan_hash']
 c=g.inspect();result['analytics']={k:c['State'].get(k) for k in ['Status','OOMKilled']}
 result['analytics']['health']=c['State'].get('Health',{}).get('Status');result['analytics']['restart_count']=c['RestartCount']
 result['env_cas_preserved']=env_after is None or g.sha(g.ENV)==env_after
 result['overlay_cas_preserved']=overlay_after is None or g.sha(overlay)==overlay_after
 result['gpu_final_mib']=g.gpu();save()
 print(json.dumps({k:result.get(k) for k in ['stage','readiness','calls','metadata_verified','sample_count','valid_bbox_observed',
  'shared_visible_track_observed','cleanup_verified','analytics','restart_restored','sample_capture_disabled',
  'other_containers_unchanged','window_error','exception_type']},indent=2))