"""Scoped L01 cache orchestration; legacy inference remains byte-identical.""" import concurrent.futures import datetime import json import os from pathlib import Path import shutil import signal import subprocess import sys import time ROOT=Path(__file__).resolve().parent.parent BASE=ROOT.parent BASELINE=BASE/'2026-10-29_contextra-self-filming-pose-cache_r01' PACKAGE=BASE/'2026-10-29_contextra-self-filming-original01-clips_r01/아이서티_컨텍스트라_싸움_쓰러짐_자체촬영_원본01_클립자료' REMOTE='/storage/contextra/experiments/action-model-eval/2026/'+ROOT.name PREFIX='contextra-l01pose-1029-r01-' sys.path.insert(0,str(BASELINE/'helpers')) from control import run, SSH, checked, resources, exclusive_gpu, digest, save, ENGINE, PROD def scp(src,dst): cmd=['scp','-q','-r','-i',SSH[2],'-o',SSH[4],'-o','StrictHostKeyChecking=yes','-o','BatchMode=yes', '-o','ConnectTimeout=10','-o','ServerAliveInterval=15','-o','ServerAliveCountMax=3','-P','1022',str(src),str(dst)] for attempt in range(2): p=subprocess.Popen(cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE,start_new_session=True) try: out,err=p.communicate(timeout=600) except subprocess.TimeoutExpired: os.killpg(p.pid,signal.SIGTERM) try:out,err=p.communicate(timeout=10) except subprocess.TimeoutExpired: os.killpg(p.pid,signal.SIGKILL);out,err=p.communicate(timeout=10) if p.returncode==0:return if attempt==0:time.sleep(3) raise RuntimeError('Bounded transfer failed; completed remote cache preserved: '+err.decode()[-1000:]) def checkpoint(stage): save(ROOT/'checkpoint.json',dict(stage=stage,goal='L01 two-part Pose/RGB/Tracking cache only', parts={'L01_P01':[1,2542],'L01_P02':[2543,19975]},expected_samples=[1044,7261], source_frame='L01 1-based',seconds='L01-relative actual PTS',sample='part-local 0-based', tracking='fresh per part; identity namespace=(part_id,T)', excluded=['Event evaluation','engine execution','new video','RTSP','existing C01-C22 changes'], next='reuse verified completed stages; failed transfer never triggers inference')) def snapshot(label): value=dict(timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), containers=checked(['docker','ps','--format','{{.Names}} {{.ID}} {{.Image}}']).decode().splitlines(), config_sha256=checked(['docker','exec',PROD,'sha256sum','/var/lib/contextra/runtime/streams_runtime.yaml']).decode().split()[0], gpu_owner=exclusive_gpu(),resources=resources(gpu=True)) save(ROOT/'reports'/('production_'+label+'.json'),value) def setup(): assert not (ROOT/'specs.json').exists() assert shutil.disk_usage('/').free>=15*1024**3 and shutil.disk_usage('/data').free>=30*1024**3 exclusive_gpu();resources(gpu=True,starting=True) old_inventory=json.loads((BASELINE/'manifest.json').read_text())['files'] for name in ('pose.py','sampling.py','verify_cache.py','verify_extra.py'): source=BASELINE/'helpers'/name assert digest(source)==old_inventory['helpers/'+name]['sha256'] assert not (ROOT/'helpers'/name).exists() shutil.copy2(source,ROOT/'helpers'/name) shutil.copy2(BASELINE/'model_reference.json',ROOT/'model_reference.json') ref=json.loads((ROOT/'model_reference.json').read_text()) assert checked(['docker','image','inspect',ref['image'],'--format','{{.Id}}']).decode().strip()==ref['image'] assert checked(['sha256sum',ENGINE+'/model.plan'],sudo=True).decode().split()[0]==ref['pose']['engine_sha256'] for name,h in ref['pose']['source_files'].items(): assert checked(['docker','exec',PROD,'sha256sum','/opt/contextra/app/'+name]).decode().split()[0]==h manifest=json.loads((PACKAGE/'metadata/manifest.json').read_text()) clip=next(x for x in manifest['clips'] if x['id']=='L01') checks=json.loads((PACKAGE/'SHA256SUMS.json').read_text())['files'] for rel in (clip['output'],'metadata/frame_maps/L01.csv'): assert digest(PACKAGE/rel)==checks[rel]['sha256'] specs={} for key,bounds,n in [('L01_P01',[1,2542],1044),('L01_P02',[2543,19975],7261)]: specs[key]=dict(local_source=str(PACKAGE/clip['output']),remote_source=REMOTE+'/sources/L01.mp4', sha256=checks[clip['output']]['sha256'],display_bounds=bounds,known_loop_boundaries=[], cache=REMOTE+'/clips/'+key,reused=False,expected_samples=n,part_id=key) save(ROOT/'specs.json',specs) save(ROOT/'reports/source_fingerprints.json',dict(input_sha256=checks[clip['output']]['sha256'], frame_map_sha256=checks['metadata/frame_maps/L01.csv']['sha256'],baseline_manifest_sha256=digest(BASELINE/'manifest.json'), baseline_catalog_sha256=digest(BASELINE/'metadata/catalog.json'),observations_sha256=digest(BASELINE/'metadata/user_observations.md'))) checkpoint('PREPARING_P01') checked(['test','!','-e',REMOTE]);checked(['mkdir',REMOTE],sudo=True);checked(['chown','si:si',REMOTE],sudo=True);checked(['chmod','700',REMOTE]) dirs=[REMOTE+'/sources',REMOTE+'/reports'] for key in specs: dirs += [REMOTE+'/clips/'+key+'/'+d for d in ('inputs','reports','pose_models/RFDETRKeypointPreview/8')] checked(['mkdir','-p']+dirs) sync() scp(specs['L01_P01']['local_source'],'si@58.233.207.31:'+REMOTE+'/sources/L01.mp4') assert checked(['sha256sum',REMOTE+'/sources/L01.mp4']).decode().split()[0]==specs['L01_P01']['sha256'] snapshot('before') print('SETUP_PASS L01 single source; no inference yet',flush=True) def sync(): scp(ROOT/'helpers','si@58.233.207.31:'+REMOTE+'/') for name in ('specs.json','model_reference.json','checkpoint.json'): scp(ROOT/name,'si@58.233.207.31:'+REMOTE+'/') def guard_gpu_owner(name): allowed=set(checked(['docker','top',PROD,'-eo','pid']).decode().split()[1:]) own=run(['docker','top',name,'-eo','pid'],timeout=30) if own.returncode==0:allowed.update(own.stdout.decode().split()[1:]) actual=set(checked(['nvidia-smi','--query-compute-apps=pid','--format=csv,noheader,nounits']).decode().split()) assert actual<=allowed,'Concurrent GPU owner detected; stop only this trial' def trial(stage,key): s=json.loads((ROOT/'specs.json').read_text())[key] ref=json.loads((ROOT/'model_reference.json').read_text());gpu=stage=='pose' resources(gpu=gpu,starting=gpu) if gpu:exclusive_gpu() name=PREFIX+stage.replace('_','-')+'-'+key.lower() 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','-e','PYTHONDONTWRITEBYTECODE=1', '-e','OMP_NUM_THREADS=2','-e','OPENBLAS_NUM_THREADS=2','-e','CLIP='+key, '-v',s['cache']+':/eval','-v',s['cache']+':/cache:ro','-v',REMOTE+'/helpers:/helpers:ro', '-v',REMOTE+'/specs.json:/specs.json:ro','-v',REMOTE+'/model_reference.json:/model_reference.json:ro', '-v',s['remote_source']+':/source.mp4:ro'] if key=='L01_P02' and stage=='plan':cmd+=['-v',REMOTE+'/clips/L01_P01:/prefix_cache:ro'] if gpu:cmd+=['--gpus','device=0','-v',ENGINE+':/eval/pose_models/RFDETRKeypointPreview/8:ro'] else:cmd+=['--runtime','runc','-e','NVIDIA_VISIBLE_DEVICES=void'] cmd+=['--entrypoint','python3',ref['image'],'/helpers/'+stage+'.py'] print('START',key,stage,flush=True);samples=[] with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: future=pool.submit(run,cmd,timeout=1800) try: while True: try:p=future.result(timeout=30);break except concurrent.futures.TimeoutError: if future.done():raise if gpu:guard_gpu_owner(name) samples.append(resources(gpu=gpu));print('MONITOR',key,stage,samples[-1].get('gpu_free_mib'),flush=True) except BaseException: run(['docker','stop','--time','15',name],timeout=40);raise (ROOT/'reports'/f'{key}_{stage}.log').write_bytes(p.stdout+p.stderr) save(ROOT/'reports'/f'{key}_{stage}_resources.json',samples) if p.returncode: run(['docker','stop','--time','15',name],timeout=40) raise RuntimeError(key+'/'+stage+': '+p.stderr.decode()[-1500:]) print('PASS',key,stage,flush=True) def infer(key): s=json.loads((ROOT/'specs.json').read_text())[key] if key=='L01_P02': first=json.loads((ROOT/'clips/L01_P01/reports/compatibility.json').read_text()) assert first['status']=='PASS' and first['samples']==1044 for stage,marker in [('plan','inputs/sampling_plan.json'),('pose','inputs/pose.json'),('verify_cache','reports/cache_validation.json'),('verify_extra','reports/compatibility.json')]: if run(['test','-f',s['cache']+'/'+marker]).returncode==0: print('REUSE',key,stage,flush=True);continue trial(stage,key) local=ROOT/'clips'/key (local/'inputs').mkdir(parents=True,exist_ok=True);(local/'reports').mkdir(exist_ok=True) remote_hashes=json.loads(checked(['cat',s['cache']+'/reports/cache_hashes.json'])) for name in ('pose.json','sampling.csv','source_map.csv','sampling_plan.json','pose_candidates_prefilter.jsonl'): rel='inputs/'+name;dest=local/rel if dest.is_file() and digest(dest)==remote_hashes['files'][rel]:continue scp('si@58.233.207.31:'+s['cache']+'/'+rel,dest) scp('si@58.233.207.31:'+s['cache']+'/reports/.',local/'reports') for rel,h in remote_hashes['files'].items(): if rel.startswith('inputs/frames/'):continue assert digest(local/rel)==h,rel assert json.loads((local/'reports/compatibility.json').read_text())['samples']==s['expected_samples'] checkpoint('P01_COMPATIBILITY_PASS' if key=='L01_P01' else 'BOTH_PART_CACHES_PASS') if key=='L01_P02':snapshot('after') print('TRANSFER_VERIFIED',key,flush=True) if __name__=='__main__': if sys.argv[1]=='setup':setup() elif sys.argv[1]=='sync':sync() elif sys.argv[1]=='first':infer('L01_P01') elif sys.argv[1]=='second':infer('L01_P02') else:raise ValueError('Unknown stage')