"""CPU-only runner; mutations are confined to this unique experiment.""" import concurrent.futures import hashlib import json from pathlib import Path import sys import time LOCAL = Path(__file__).resolve().parent.parent SOURCE = LOCAL.parent / '2026-10-28_contextra-pose-input-review-15clips_r01' AB = LOCAL.parent / '2026-10-28_contextra-fight-ab-6clips_r01' sys.path.insert(0, str(SOURCE / 'helpers')) import control as old REMOTE = old.RB + '/' + LOCAL.name PREVIOUS = old.REMOTE + '/download' REMOTE_AB = old.RB + '/' + AB.name PREFIX = 'contextra-bdiag-1029-r01-' REPORT = LOCAL / 'reports' def save(name, value): (REPORT / name).write_text(json.dumps(value, ensure_ascii=False, indent=2)) def resources(): for attempt in range(3): try: return old.resources() except RuntimeError as exc: if 'banner exchange' not in str(exc) or attempt==2: raise print('RETRY_READONLY_RESOURCE',attempt+1,flush=True) time.sleep(2) def launch(cmd, stage, clip): for attempt in range(3): result=old.run(cmd,timeout=1800) if result.returncode!=255 or b'banner exchange' not in result.stderr: return result (REPORT/(stage+'_'+clip+f'_preconnect_attempt{attempt+1}.log')).write_bytes(result.stderr) if attempt<2:time.sleep(2) return result def snapshot(label): state = json.loads(old.checked(['docker', 'exec', '-i', 'contextra-ai2-analytics-1', 'python3', '-'], data=(old.PREP / 'helpers/snapshot.py').read_bytes(), timeout=90)) assert state['health']['ready'] save('production_' + label + '.json', state) entries = old.checked(['docker', 'ps', '--format', '{{.Names}} {{.Image}} {{.ID}}']).decode().splitlines() save('containers_' + label + '.json', entries) return state def sync(): old.checked(['mkdir','-p',REMOTE+'/reports/partials']) for name in ('helpers', 'specs.json', 'baseline_hashes.json', 'checkpoint.ko.md'): old.scp(str(LOCAL / name), 'si@58.233.207.31:' + REMOTE + '/') def setup(): REPORT.mkdir(exist_ok=True) old.checked(['test', '!', '-e', REMOTE]) save('resources_before.json', old.resources()) assert old.checked(['docker', 'image', 'inspect', old.IMAGE, '--format', '{{.Id}}']).decode().strip() == old.IMAGE specs = json.loads((SOURCE / 'specs.json').read_text()) assert len(specs) == 15 and len({s['sha256'] for s in specs.values()}) == 15 previous = SOURCE / 'Contextra_Pose_Input_Review_20261028' inventory = json.loads((previous / 'SHA256SUMS.json').read_text()) for clip in specs: for name in ('cache_validation.json', 'identity_palette.json', 'input_rgb_sha256.json', 'frame_map.csv'): rel = clip + '/data/' + name assert hashlib.sha256((previous / rel).read_bytes()).hexdigest() == inventory[rel] (LOCAL / 'specs.json').write_bytes((SOURCE / 'specs.json').read_bytes()) names = ('features.py', 'state.py', 'engines.py', 'layout.py', 'contract.py') hashes = {name: hashlib.sha256((AB / 'helpers' / name).read_bytes()).hexdigest() for name in names} for name, digest in hashes.items(): assert old.checked(['sha256sum', REMOTE_AB + '/helpers/' + name]).decode().split()[0] == digest (LOCAL / 'baseline_hashes.json').write_text(json.dumps(hashes, indent=2)) old.checked(['mkdir', REMOTE], sudo=True) old.checked(['chown', 'si:si', REMOTE], sudo=True) old.checked(['chmod', '700', REMOTE]) old.checked(['mkdir', '-p', REMOTE + '/download/validation']) sync() snapshot('before') print('SETUP_PASS input=15 GPU=0 baseline_read_only=1', flush=True) def trial(stage, clip): resources() spec = json.loads((LOCAL / 'specs.json').read_text())[clip] name = PREFIX + stage + '-' + clip[:2] 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 + '/helpers:/helpers:ro', '-v', REMOTE + '/specs.json:/specs.json:ro', '-v', REMOTE + '/baseline_hashes.json:/baseline_hashes.json:ro', '-v', REMOTE_AB + '/helpers:/baseline_helpers:ro', '-v', REMOTE_AB + '/delivery:/ab_previous:ro', '-v', spec['cache'] + ':/cache:ro', '-v', PREVIOUS + ':/previous:ro', '-v', REMOTE + '/download:/download', '-v', REMOTE + '/reports/partials:/partials', '-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 = [] with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(launch, cmd, stage, clip) try: while True: try: result = future.result(timeout=30) break except concurrent.futures.TimeoutError: if future.done(): raise samples.append(resources()) print('RESOURCE', stage, clip, len(samples), flush=True) except BaseException: old.run(['docker', 'stop', '--time', '15', name], timeout=40) raise (REPORT / (stage + '_' + clip + '.log')).write_bytes(result.stdout + result.stderr) save(stage + '_' + clip + '_resources.json', samples) if result.returncode: old.run(['docker','stop','--time','15',name],timeout=40) raise RuntimeError(result.stderr.decode()[-2000:] + result.stdout.decode()[-1000:]) print('DONE', stage, clip, result.stdout.decode()[-1000:], flush=True) def finish(): after = snapshot('after') before = json.loads((REPORT / 'production_before.json').read_text()) fields = ('state', 'run_id', 'run_generation', 'child_pid', 'cleanup_pending', 'last_error') changes = {k: {f: [before['streams'].get(k, {}).get(f), v.get(f)] for f in fields if before['streams'].get(k, {}).get(f) != v.get(f)} for k, v in after['streams'].items()} remaining = old.checked(['docker', 'ps', '-a', '--filter', 'name=' + PREFIX, '--format', '{{.Names}}']).decode().strip() assert not remaining def production(items): return sorted(s for s in items if s.startswith('contextra-ai2-')) result = dict(config_model_unchanged={k: before[k] == after[k] for k in ('config_sha256', 'pose_engine_sha256', 'pose_config_sha256')}, containers_unchanged=production(json.loads((REPORT / 'containers_before.json').read_text())) == production(json.loads((REPORT / 'containers_after.json').read_text())), stream_changes={k: v for k, v in changes.items() if v}, trial_containers_remaining=[], production_control_calls=0, gpu_jobs=0, unrelated_trials='Observed only; never stopped or changed') save('operational_comparison.json', result) save('resources_after.json', old.resources()) 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 == 'one': trial(sys.argv[2], sys.argv[3]) elif stage in ('evaluate_review', 'render_review', 'audit_review'): clips = list(json.loads((LOCAL / 'specs.json').read_text())) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: for future in concurrent.futures.as_completed([pool.submit(trial, stage, c) for c in clips]): future.result() else: raise ValueError(stage)