from gate import R,REPO,ENV,CFG,HEAD,sha,shell,inspect_all,other_baseline import json,os,tempfile,pathlib,subprocess,time,urllib.request,urllib.parse,datetime import yaml from tools.validate_deployment import load_env_file,validate_deployment_inputs from model_tooling.deploy import deployment_plan from model_tooling.medium_acceptance import POLICY os.umask(0o077) def save(): fd,name=tempfile.mkstemp(dir=R) with os.fdopen(fd,'w') as f: json.dump(result,f,indent=2);f.flush();os.fsync(f.fileno()) os.replace(name,R/'runtime.json') def inspect(): return next(x for x in inspect_all() if x['Name']=='/contextra-analytics-1') def api(path,method='GET',timeout=5): with urllib.request.urlopen(urllib.request.Request(base+path,method=method),timeout=timeout) as f:return json.load(f) def gpu():return int(shell(['nvidia-smi','--id=0','--query-gpu=memory.used','--format=csv,noheader,nounits']).strip()) def threads(): code='''import pathlib,json counts={"processes":0,"tasks":0,"media":0} for p in pathlib.Path('/proc').glob('[0-9]*'): try: args=(p/'cmdline').read_bytes().split(bytes([0])) if len(args)<3 or args[1:3]!=[b'-m',b'app.main']:continue counts['processes']+=1 for q in (p/'task').iterdir(): name=(q/'comm').read_text().lower();counts['tasks']+=1 if any(k in name for k in ['rtp','rtsp','infer','preprocess','appsink','nvvideoconvert']):counts['media']+=1 except (FileNotFoundError,ProcessLookupError):pass print(json.dumps(counts))''' return json.loads(shell(['docker','exec','contextra-analytics-1','python3','-c',code])) def stop_analytics(): subprocess.run(['docker','stop','--timeout','20','contextra-analytics-1'],check=True,capture_output=True,timeout=35) result['analytics_safety_stop']=True;save() def safe_state(s): return {k:s.get(k) for k in ['state','desired_state','observed_state','run_generation','cleanup_pending','cleanup_error','last_error','per_task_buffers']} assert not (R/'runtime.json').exists() gate=json.loads((R/'gate.json').read_text());build=json.loads((R/'build.json').read_text()) assert json.loads((R/'image-verify.json').read_text())['status']=='PASS' assert shell(['git','-C',str(REPO),'rev-parse','HEAD']).strip()==HEAD assert not shell(['git','-C',str(REPO),'status','--porcelain']).strip() assert sha(ENV)==gate['env_hash'] and sha(CFG)==gate['config_hash'] assert sha(CFG.parent/'2/model.plan')==gate['plan_hash'] c=inspect();assert c['Image']==gate['old_image'] and c['State']['Status']=='exited' env=load_env_file(str(ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml' assert sha(overlay)==gate['overlay_hash'] assert other_baseline(inspect_all())==gate['other_containers'] 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' s=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'] result={'stage':'GATE_PASS','start_attempts':0,'stop_attempts':0,'source':HEAD,'image':build['image_id'],'version1_rollback':False} save();changed=False;since=None try: lines=ENV.read_text().splitlines(keepends=True);matches=[] for i,line in enumerate(lines): if line.split('=',1)[0].strip()=='CONTEXTRA_IMAGE':matches.append(i) assert len(matches)==1 lines[matches[0]]='CONTEXTRA_IMAGE='+build['tag']+'\n' assert sha(ENV)==gate['env_hash'] fd,name=tempfile.mkstemp(dir=ENV.parent) with os.fdopen(fd,'w') as f:os.fchmod(f.fileno(),0o600);f.write(''.join(lines));f.flush();os.fsync(f.fileno()) os.replace(name,ENV);changed=True result.update(stage='ENV_APPLIED',env_after_hash=sha(ENV));save() validate_deployment_inputs(str(ENV)) compose=['docker','compose','--env-file',str(ENV),'-f',str(REPO/'docker-compose.yml')] subprocess.run(compose+['config','--quiet'],check=True,capture_output=True,timeout=30) render=json.loads(subprocess.check_output(compose+['config','--format','json'],text=True,timeout=30)) assert render['services']['analytics']['image']==build['tag'] with (R/'compose.log').open('w') as log: subprocess.run(compose+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],check=True,stdout=log,stderr=subprocess.STDOUT,timeout=120) end=time.monotonic()+300;ready=False while time.monotonic()8192:result['budget_exceeded']=True;break if state.get('last_error'):break if frames and state['observed_state']=='running': if first is None:first=time.monotonic() if frames>=100 or time.monotonic()-first>=10:success=True;break time.sleep(.5) finally: result['stop_attempts']=1;save() try: api(path+'/stop','POST',25) end=time.monotonic()+10 while True: post=api(path);result['post_stop']=safe_state(post) if not post.get('cleanup_pending') or time.monotonic()>end:break time.sleep(.5) result['threads_after']=threads() cleaned=(all(post[k]=='stopped' for k in ['state','desired_state','observed_state']) and not post.get('cleanup_pending') and not post.get('cleanup_error') and result['threads_after']['media']==0) result['cleanup_verified']=cleaned if not cleaned:stop_analytics() except Exception as error: result['cleanup_verified']=False;result['cleanup_exception']=type(error).__name__;stop_analytics() result['window_seconds']=round(time.monotonic()-started,2) log=subprocess.run(['docker','logs','--since',since,'--tail','250','contextra-analytics-1'],capture_output=True,text=True,timeout=15) result['geometry_samples']=[];result['diagnostics']=[] for line in (log.stdout+log.stderr).splitlines(): for prefix,key in [('[NATIVE_BRIDGE_SAMPLE] ','geometry_samples'),('[NATIVE_METADATA_DIAGNOSTIC] ','diagnostics')]: if prefix in line and len(result[key])<3:result[key].append(json.loads(line.split(prefix,1)[1])) for sample in result['geometry_samples']: assert min(sample['source']+sample['surface'])>0 for d in sample['detections']: box=d['bbox_xyxy'];assert 0<=d['class_id']<80 and 0<=d['score']<=1 assert len(box)==4 and all(0<=v<=1 for v in box) and box[0]