"""One approved sequential batch; no channel controls, retries, or visual analysis.""" import ast import csv import hashlib import json import pathlib import re import shutil import subprocess import sys OLD = pathlib.Path('/data/codex-artifacts/contextra/action-model-eval/2026/2026-10-27_contextra-camera27-prep_r01/helpers') sys.path.insert(0, '/data/codex-artifacts/contextra/action-model-eval/2026/2026-10-26_contextra-camera24-prep_r01/helpers') from remote import run, SSH LOCAL = pathlib.Path(__file__).resolve().parent.parent REMOTE = '/storage/contextra/experiments/action-model-eval/2026/' + LOCAL.name TARGETS = [(22, 'crowd'), (23, 'loitering'), (26, 'intrusion'), (29, 'fire')] IMAGE = 'sha256:671c7416f503481d2be2092a5aa0c035938a07fc514cfc0b45f94d78e86370bd' ANALYTICS = 'contextra-ai2-analytics-1' CONTAINERS = [ANALYTICS, 'contextra-ai2-ai-control-sidecar-1', 'contextra-ai2-ai-gateway-1', 'contextra-ai2-redis-1', 'contextra-ai2-postgresql-1'] SCP = ['scp'] + SSH[1:-3] + ['-P', SSH[-2]] def command(args, sudo=False, timeout=45): result = run(args, sudo=sudo, timeout=timeout) if result.returncode: raise RuntimeError('bounded remote command failed; raw stderr withheld') return result.stdout def save(path, value): with path.open('x') as handle: json.dump(value, handle, indent=2, ensure_ascii=False) def transfer(sources, dest, timeout=240): result = subprocess.run(SCP + sources + [dest], capture_output=True, timeout=timeout) if result.returncode: raise RuntimeError('bounded transfer failed; raw stderr withheld') SNAPSHOT_CODE = """import datetime,hashlib,json,pathlib,urllib.request,yaml raw=pathlib.Path('/var/lib/contextra/runtime/streams_runtime.yaml').read_bytes() rows=yaml.safe_load(raw)['added']; rows=list(rows.values()) if isinstance(rows,dict) else rows targets={22:'crowd',23:'loitering',26:'intrusion',29:'fire'}; selected={} for cid,scenario in targets.items(): matches=[s for s in rows if str(s.get('camera_id',s.get('source_id')))==str(cid)] if len(matches)!=1: selected[str(cid)]={'registration_count':len(matches)};continue s=matches[0]; config=s.get(scenario) selected[str(cid)]={'registration_count':1,'camera_id':cid,'stream_id':s.get('stream_id'),'expected_scenario':scenario,'scenario_config':config,'analysis_fps_target':s.get('analysis_fps_target'),'source_present':bool(s.get('uri')),'source_sha256':hashlib.sha256(str(s.get('uri','')).encode()).hexdigest()} with urllib.request.urlopen('http://127.0.0.1:5000/api/v1/streams?detail=true',timeout=15) as r: streams=json.load(r)['streams'] print(json.dumps({'at':datetime.datetime.now(datetime.timezone.utc).isoformat(),'config_sha256':hashlib.sha256(raw).hexdigest(),'targets':selected,'streams':[{k:x.get(k) for k in ('stream_id','state','run_id','run_generation','child_pid','cleanup_pending')} for x in streams]})) """ def snapshot(): data = json.loads(command(['docker', 'exec', ANALYTICS, 'python3', '-c', SNAPSHOT_CODE], sudo=True)) data['containers'] = command(['docker', 'inspect', *CONTAINERS, '--format', '{{json .Name}} {{json .Image}} {{json .State.StartedAt}}'], sudo=True).decode().splitlines() return data def safety(): assert shutil.disk_usage('/').free >= 15 * 1024**3, 'Local root disk safety boundary' assert shutil.disk_usage(LOCAL).free >= 15 * 1024**3, 'Local output disk safety boundary' code = "import json,shutil;print(json.dumps({p:shutil.disk_usage(p).free for p in ['/', '/storage']}))" free = json.loads(command(['python3', '-B', '-c', code])) assert min(free.values()) >= 15 * 1024**3, 'Remote disk safety boundary' return free def container(cid, phase): name = f'contextra-{phase}-camera{cid}-20261027-batch-r01' assert not command(['docker','ps','-aq','--filter','name=^/'+name+'$'], sudo=True).strip(), 'Container name already in use' base = ['docker','run','--rm','--pull','never','--name',name,'--runtime','runc','--network', 'host' if phase=='record' else 'none', '--user','0:0','--read-only','--tmpfs','/tmp:rw,nosuid,size=128m','--cap-drop','ALL','--cap-add','DAC_OVERRIDE', '--security-opt','no-new-privileges','--cpus','2','--memory','2g','--pids-limit','128','-e','NVIDIA_VISIBLE_DEVICES=void', '-e','PYTHONDONTWRITEBYTECODE=1','--mount',f'type=bind,src={REMOTE}/camera{cid},dst=/work'] if phase=='record': base += ['--mount','type=bind,src=/storage/contextra/runtime,dst=/source,readonly'] args = ['--camera-id',str(cid)] if phase=='record': args += ['--seconds','120','--scenario',dict(TARGETS)[cid]] base += ['--entrypoint','python3',IMAGE,'/work/helpers/'+('capture.py' if phase=='record' else 'review_video.py'),*args] print(f'camera{cid}: {phase} started', flush=True) try: result = run(base,sudo=True,timeout=220 if phase=='record' else 1100) log = re.sub(r'[a-zA-Z]+://[^\s]+','',(result.stdout+result.stderr).decode(errors='replace')) save(LOCAL/f'camera{cid}/reports/{phase}_execution.json', {'returncode':result.returncode,'output':log[-12000:]}) return result.returncode finally: if command(['docker','ps','-aq','--filter','name=^/'+name+'$'],sudo=True).strip(): command(['docker','stop','--time','10',name],sudo=True,timeout=30) assert not command(['docker','ps','-aq','--filter','name=^/'+name+'$'],sudo=True).strip(), 'Trial cleanup failed' def main(): reports=LOCAL/'reports'; reports.mkdir(exist_ok=True) assert not (reports/'operational_before.json').exists(), 'Existing batch cannot be rerun' safety() assert command(['docker','image','inspect',IMAGE,'--format','{{.Id}}'],sudo=True).decode().strip()==IMAGE mounts=json.loads(command(['docker','inspect',ANALYTICS,'--format','{{json .Mounts}}'],sudo=True)) assert any(m['Source']=='/storage/contextra/runtime' and m['Destination']=='/var/lib/contextra/runtime' for m in mounts) toolcode="import shutil,pathlib,yaml;assert shutil.which('ffmpeg') and shutil.which('ffprobe');assert pathlib.Path('/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf').is_file();print('TOOLS_READY')" print(command(['docker','run','--rm','--pull','never','--runtime','runc','--network','none','--read-only','--cpus','1','--memory','512m','--pids-limit','64','-e','NVIDIA_VISIBLE_DEVICES=void','--entrypoint','python3',IMAGE,'-B','-c',toolcode],sudo=True).decode(),flush=True) before=snapshot();save(reports/'operational_before.json',before) command(['mkdir','-m','700',REMOTE],sudo=True); command(['chown','si:si',REMOTE],sudo=True) command(['mkdir',REMOTE+'/reports']) results=[] try: for cid,scenario in TARGETS: safety() current=snapshot(); target=current['targets'][str(cid)] if current['containers']!=before['containers']: results.append({'camera_id':cid,'status':'DEFERRED_COMMON_SERVICE_DRIFT'});break config=target.get('scenario_config') if not isinstance(config,dict) or config.get('enabled') is not True or not target.get('source_present') or (scenario=='fire' and config.get('scenario',{}).get('enabled') is not True): results.append({'camera_id':cid,'status':'DEFERRED_REGISTRATION'});continue loc=LOCAL/f'camera{cid}'; loc.mkdir(); for folder in ('helpers','inputs','reports'): (loc/folder).mkdir() save(loc/'reports/registration.json',{'config_sha256':current['config_sha256'],**target}) command(['mkdir',REMOTE+f'/camera{cid}',REMOTE+f'/camera{cid}/helpers',REMOTE+f'/camera{cid}/inputs',REMOTE+f'/camera{cid}/reports']) for helper in ('capture.py','review_video.py'): ast.parse((OLD/helper).read_text()); shutil.copy2(OLD/helper,loc/'helpers'/helper) transfer([str(loc/'helpers/capture.py'),str(loc/'helpers/review_video.py')],SSH[-1]+':'+REMOTE+f'/camera{cid}/helpers/') code=container(cid,'record') if code: results.append({'camera_id':cid,'status':'DEFERRED_CAPTURE','returncode':code}); if code in (137,139): break continue code=container(cid,'review') if code: results.append({'camera_id':cid,'status':'DEFERRED_REVIEW','returncode':code}); if code in (137,139): break continue for folder,names in [('inputs',[f'camera{cid}_original.mp4',f'camera{cid}_review.mp4',f'camera{cid}_frame_map.csv']),('reports',['capture.json','original_probe.json','review_probe.json','video_validation.json'])]: transfer([SSH[-1]+':'+REMOTE+f'/camera{cid}/{folder}/'+n for n in names],str(loc/folder)+'/') v=json.loads((loc/'reports/video_validation.json').read_text()) for path in (loc/'inputs').iterdir(): digest=hashlib.sha256() with path.open('rb') as f: for chunk in iter(lambda:f.read(1048576),b''): digest.update(chunk) assert digest.hexdigest()==v[path.name]['sha256'] and path.stat().st_size==v[path.name]['bytes'] with (loc/f'inputs/camera{cid}_frame_map.csv').open(newline='') as f: rows=list(csv.DictReader(f)) assert [int(r['frame_1based']) for r in rows]==list(range(1,v['frame_count']+1)) assert max(abs(float(r['elapsed_seconds'])-float(r['review_pts_seconds'])) for r in rows)<0.001 save(loc/'reports/transfer_validation.json',{'files':3,'size_sha256_match':True,'frame_map_rows':len(rows),'sequential':True}) results.append({'camera_id':cid,'scenario':scenario,'status':'READY','frames':v['frame_count'],'elapsed_seconds':v['elapsed_seconds'],'fps':v['original_stream']['r_frame_rate']}) print(json.dumps(results[-1]),flush=True) finally: for cid,_ in TARGETS: if cid not in [x['camera_id'] for x in results]: results.append({'camera_id':cid,'status':'NOT_RUN'}) save(reports/'batch_results.json',results) after=snapshot(); save(reports/'operational_after.json',after) a={s['stream_id']:s for s in before['streams']}; b={s['stream_id']:s for s in after['streams']} comparison={k:before[k]==after[k] for k in ['config_sha256','targets','containers']} comparison['channel_state_changes']=[{'stream_id':k,'before':a.get(k),'after':b.get(k)} for k in sorted(set(a)|set(b)) if a.get(k)!=b.get(k)] comparison['running_before']=sum(s['state']=='running' for s in a.values()); comparison['running_after']=sum(s['state']=='running' for s in b.values()) comparison['temporary_containers_remaining']=command(['docker','ps','-aq','--filter','name=^/contextra-(record|review)-camera(22|23|26|29)-20261027-batch-r01$'],sudo=True).decode().splitlines() save(reports/'operational_comparison.json',comparison) print(json.dumps({'batch_results':results,'comparison':comparison}),flush=True) if __name__=='__main__': main()