"""Bounded CPU trials. No production control or model inference.""" import concurrent.futures import csv import hashlib import json from pathlib import Path import subprocess import sys import time LOCAL = Path(__file__).resolve().parent.parent BASE = LOCAL.parent SOURCE = BASE / '2026-10-28_contextra-pose-input-review-15clips_r01' AB = BASE / '2026-10-28_contextra-fight-ab-6clips_r01' BDIAG = BASE / '2026-10-29_contextra-b-diagnostics-15clips_r01' SELF = BASE / '2026-10-29_contextra-self-filming-pose-cache_r01' UNIFIED = BASE / '2026-10-29_contextra-self-filming-l01-pose-cache_r01' sys.path.insert(0, str(SOURCE / 'helpers')) import control as old REMOTE = old.RB + '/' + LOCAL.name PREFIX = 'contextra-brefine-1029-r01-' REPORT = LOCAL / 'reports' DEVELOP = ['01_camera24_fight', '02_camera22_crowd', '03_camera27_fall', '07_KISA_fight_C00_015_0009', '14_KISA_fall_C00_051_0002'] DEVELOP += ['SELF_' + x for x in ('C05','C09','C10','C13','C14','C15','C17','C21')] HOLDOUT = ['08_KISA_fight_C00_018_0012', '09_KISA_fight_C00_021_0011'] def read(path): return json.loads(Path(path).read_text()) def save(name, value): (REPORT / name).write_text(json.dumps(value, ensure_ascii=False, indent=2)) def digest(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def snapshot(label): names = old.checked(['docker','ps','--format','{{.Names}}']).decode().splitlines() names = [n for n in names if not n.startswith(PREFIX)] fmt = '{{.Name}}|{{.Image}}|{{.State.Status}}|{{.State.StartedAt}}|{{.RestartCount}}|{{if .State.Health}}{{.State.Health.Status}}{{else}}not-configured{{end}}' containers = old.checked(['docker','inspect','--format',fmt]+names).decode().splitlines() if names else [] gpu = old.checked(['nvidia-smi','--query-compute-apps=pid,process_name,used_memory','--format=csv,noheader'],timeout=30).decode().splitlines() state = dict(containers=sorted(containers), gpu_processes=sorted(gpu), resources=old.resources(), application_channel_health='NOT_MEASURED',production_control_calls=0) save('production_'+label+'.json',state) return state def setup(): REPORT.mkdir(exist_ok=True) old.checked(['test','!','-e',REMOTE]) assert old.checked(['docker','image','inspect',old.IMAGE,'--format','{{.Id}}']).decode().strip()==old.IMAGE specs={} previous=SOURCE/'Contextra_Pose_Input_Review_20261028' for name,s in read(SOURCE/'specs.json').items(): data=previous/name/'data'; check=read(data/'cache_validation.json') assert check['status']=='PASS' specs[name]=dict(cache=s['cache'],cache_sha256=check['cache_sha256'], selected_samples=check['selected_samples'],rgb_hashes=str(data/'input_rgb_sha256.json'), rgb_hashes_remote=old.RB+'/'+SOURCE.name+'/download/'+name+'/data/input_rgb_sha256.json', input_sha256=s['sha256'],source_file=s['source_file'],domain='legacy', baseline=old.RB+'/'+BDIAG.name+'/download/'+name+'/data/B_diagnostics.jsonl') catalog=read(SELF/'metadata/catalog.json') unified=read(UNIFIED/'metadata/catalog.json') assert len(catalog['clips'])==22 and unified['L01']['observations']==8305 maps={} for c in catalog['clips']: name='SELF_'+c['clip_id']; check=read(SELF/c['reader_validation']) assert check['status']=='PASS' with (SELF/c['observation_map']).open(encoding='utf-8-sig',newline='') as f: maps[name]=[{k:r[k] for k in ('sample','clip_frame_1based','clip_pts_seconds','review_frame_1based','review_pts_seconds','region')} for r in csv.DictReader(f)] specs[name]=dict(cache=old.RB+'/'+SELF.name+'/'+c['cache'],cache_sha256=check['cache_sha256'], selected_samples=check['selected_samples'],rgb_hashes=str(SELF/c['reader_rgb_hashes']), rgb_hashes_remote=old.RB+'/'+SELF.name+'/'+c['reader_rgb_hashes'], input_sha256=c['video']['sha256'],source_file=c['video']['path'],domain='self', clip_id=c['clip_id'],title=c['name'],baseline=None) assert len(specs)==37 (LOCAL/'specs.json').write_text(json.dumps(specs,ensure_ascii=False,indent=2)) (LOCAL/'self_maps.json').write_text(json.dumps(maps,ensure_ascii=False,separators=(',',':'))) (LOCAL/'cases.json').write_bytes((SELF/'metadata/action_cases.json').read_bytes()) contract=dict(image=old.IMAGE,cpus=2,memory_gib=8,max_workers=2,timeout_seconds=1800, root_min_gib=15,storage_min_gib=30,ram_min_gib=12,gpu_jobs=0, development=DEVELOP,holdout=HOLDOUT,L01='DEFERRED',source_baseline_sha256=digest(AB/'helpers/engines.py'), input_catalog_sha256=digest(UNIFIED/'metadata/catalog.json'),retention='manual-review', operating_application='Changed before task; do not assume legacy container names or health') (LOCAL/'execution_contract.json').write_text(json.dumps(contract,indent=2)) snapshot('before') old.checked(['mkdir',REMOTE],sudo=True) old.checked(['chown','si:si',REMOTE],sudo=True) old.checked(['chmod','700',REMOTE]) old.checked(['mkdir','-p',REMOTE+'/work',REMOTE+'/download/validation',REMOTE+'/reports']) sync() print('SETUP_PASS clips=37 L01=deferred GPU=0',flush=True) def sync(): for name in ('helpers','specs.json','cases.json','self_maps.json','execution_contract.json','checkpoint.ko.md'): old.scp(str(LOCAL/name),'si@58.233.207.31:'+REMOTE+'/') for p in sorted(LOCAL.glob('candidate*.json')): old.scp(str(p),'si@58.233.207.31:'+REMOTE+'/') if (LOCAL/'00_README_ko.txt').exists(): old.scp(str(LOCAL/'00_README_ko.txt'),'si@58.233.207.31:'+REMOTE+'/') def checked_resources(): for attempt in range(3): try:return old.resources() except RuntimeError as error: if 'banner exchange' not in str(error) or attempt==2:raise time.sleep(2) def run_job(stage, clip='all'): checked_resources() suffix=clip.lower().replace('_','-') name=PREFIX+stage+'-'+suffix cmd=['docker','run','--rm','--pull','never','--name',name,'--user','0:0','--read-only', '--network','none','--cap-drop','ALL','--cap-add','DAC_OVERRIDE','--security-opt','no-new-privileges', '--cpus','2','--memory','8g','--pids-limit','256','--shm-size','256m','--tmpfs','/tmp:rw,size=512m', '--runtime','runc','-e','NVIDIA_VISIBLE_DEVICES=void','-e','PYTHONDONTWRITEBYTECODE=1', '-e','OMP_NUM_THREADS=2','-e','OPENBLAS_NUM_THREADS=2','-e','CLIP='+clip, '-v',REMOTE+':/out','-v',REMOTE+'/helpers:/helpers:ro','-v',old.RB+':/evidence:ro', '-v',old.RB+'/'+AB.name+'/helpers:/baseline:ro', '-v',old.RB+'/'+old.SHORT.name+'/fonts/NotoSansCJK-Regular.ttc:/font.ttc:ro', '--entrypoint','python3',old.IMAGE,'/helpers/'+stage+'.py'] print('START',stage,clip,flush=True); samples=[] def launch(): for attempt in range(3): response=old.run(cmd,timeout=1800) if response.returncode!=255 or b'banner exchange' not in response.stderr:return response (REPORT/(stage+'_'+clip+'_preconnect_'+str(time.time_ns())+'.log')).write_bytes(response.stderr) if attempt<2:time.sleep(2) return response with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: f=pool.submit(launch) try: while True: try:r=f.result(timeout=30);break except concurrent.futures.TimeoutError: if f.done():raise sample=checked_resources() try: sample['trial_stats']=old.checked(['docker','stats','--no-stream','--format','{{json .}}',name],timeout=20).decode().strip() except RuntimeError: sample['trial_stats']='unavailable (job may already have completed)' samples.append(sample);print('RESOURCE',stage,clip,len(samples),flush=True) except BaseException: old.run(['docker','stop','--time','15',name],timeout=40);raise log=REPORT/(stage+'_'+clip+'.log') if log.exists():log=REPORT/(stage+'_'+clip+'_'+str(time.time_ns())+'.log') log.write_bytes(r.stdout+r.stderr) save(stage+'_'+clip+'_resources.json',samples) if r.returncode: old.run(['docker','stop','--time','15',name],timeout=40) raise RuntimeError((r.stderr+r.stdout).decode()[-3500:]) print('DONE',stage,clip,r.stdout.decode()[-1400:],flush=True) def many(stage, group): specs=read(LOCAL/'specs.json') names=DEVELOP if group=='develop' else HOLDOUT if group=='holdout' else read(LOCAL/'candidates.json')['representative_clips'] if group=='representative' else list(specs) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: for f in concurrent.futures.as_completed([pool.submit(run_job,stage,n) for n in names]):f.result() def finish(): after=snapshot('after');before=read(REPORT/'production_before.json') remaining=old.checked(['docker','ps','-a','--filter','name='+PREFIX,'--format','{{.Names}}']).decode().strip() assert not remaining result=dict(containers_unchanged=after['containers']==before['containers'], gpu_process_set_unchanged=[x.split(',')[:2] for x in after['gpu_processes']]==[x.split(',')[:2] for x in before['gpu_processes']], trial_containers_remaining=[],gpu_jobs=0,production_control_calls=0, channel_latency_event_quality='NOT_MEASURED') save('operational_comparison.json',result);print(json.dumps(result),flush=True) if __name__=='__main__': stage=sys.argv[1] if stage=='setup':setup() elif stage=='sync':sync() elif stage=='finish':finish() elif stage=='many':many(sys.argv[2],sys.argv[3]) elif stage=='one':run_job(sys.argv[2],sys.argv[3] if len(sys.argv)>3 else 'all') else:raise ValueError(stage)