import csv import hashlib import json import os from pathlib import Path import subprocess import sys import time os.umask(0o077) OLD = Path('/tmp/codex/ltb1021/2026-09-16_contextra-pose-batch_r01') sys.path.insert(0, str(OLD)) import runtime as rt import yaml R = Path(__file__).parent R.chmod(0o700) p = rt.p p.R = R rt.R = R os.chdir(rt.REPO) IMAGE = json.loads((OLD / 'image.json').read_text()) ROWS = [dict(source_id=src, stream_id=sid) for src, sid in p.CHANNELS.items()] COMPOSE = ['docker', 'compose', '--env-file', str(p.ENV), '-f', str(rt.REPO / 'docker-compose.yml'), '-f', str(p.ACTIVE)] def model_configs(): code = "import json,urllib.request; print(json.dumps({n:json.load(urllib.request.urlopen('http://127.0.0.1:8000/v2/models/'+n+'/config',timeout=5)) for n in ['RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke']}))" return json.loads(p.cmd(['docker', 'exec', 'contextra-analytics-1', 'python3', '-c', code], timeout=25)) def gate(): assert not (R / 'plan.json').exists(), 'GATE_EXISTS' assert p.cmd(['git', 'rev-parse', 'HEAD']).strip() == '19c6353bf96f6caba9a0d377bdecf0ad7a0a7368' assert not p.cmd(['git', 'status', '--porcelain']).strip(), 'DIRTY_WORKTREE' p.gate() plan = p.read('plan.json') all_states = rt.states() assert all(s in rt.ORIGINAL or p.quiescent(v) for s, v in all_states.items()), 'OTHER_ACTIVE_CHANNEL' for s in rt.ORIGINAL: q = all_states[s] assert q['inference_profile']['active_models'] == {'det': 'RFDETRMedium'} cfg = p.api('/api/v1/streams/' + s + '/settings')['configured'] assert not cfg['tracking']['enabled'] and not cfg['intrusion']['enabled'] assert cfg['analysisFpsTarget'] == 10 and not cfg['pose']['enabled'] and not cfg['fire']['enabled'] assert p.base.inspect('contextra-analytics-1')['Image'] != IMAGE['image'] inspected = json.loads(p.cmd(['docker', 'image', 'inspect', IMAGE['image']]))[0] assert inspected['Id'] == IMAGE['image'] for name, digest in IMAGE['hashes'].items(): assert p.sha(rt.REPO / name) == digest, 'SOURCE_DRIFT' plan['configs'] = model_configs() assert plan['configs']['RFDETRMedium']['instance_group'][0]['count'] == 1 assert plan['configs']['RFDETRMedium']['max_batch_size'] == 1 plan['caps'] = dict(total_seconds=1200, runtime_window_seconds=20, warmup_seconds=5, helper_count=1, receiver_count=0, helper_log_mib=4, min_free_disk_gib=15, analytics_ram_gib=16, min_host_available_gib=8, gpu_mib=40000, stop_timeout_seconds=90, first_frame_timeout_seconds=90, readiness_seconds=120) plan['restore'] = 'Always restore exact original image/Compose/env and three original running channels; no model/config/source changes.' plan['deadline'] = time.time() + 1200 plan['candidate_image'] = IMAGE['image'] plan['all_states'] = all_states p.save('plan.json', plan) rt.resource() print('GATE PASS: original Medium3 running; image-only trial; model configs preserved', flush=True) def window(label, seconds): rt.PHASE = label samples = [rt.snapshot(ROWS)] deadline = time.monotonic() + seconds while time.monotonic() < deadline: time.sleep(min(5, max(0, deadline - time.monotonic()))) samples.append(rt.snapshot(ROWS)) p.save(label + '-window.json', dict(samples=samples, complete=False)) assert all(q['state'] == 'running' and not q['error'] for q in samples[-1]['channels'].values()) a, b = samples[0], samples[-1] dt = b['at'] - a['at'] rows = [] for sid in rt.ORIGINAL: x, y = a['channels'][sid], b['channels'][sid] assert x['run_id'] == y['run_id'] and x['generation'] == y['generation'] redis_fps = None if not x['redis'].get('missing') and not y['redis'].get('missing'): redis_fps = (y['redis']['entries'] - x['redis']['entries']) / (y['redis']['at'] - x['redis']['at']) rows.append(dict(phase=label, stream_id=sid, seconds=dt, result_fps=(y['processed'] - x['processed']) / dt, redis_fps=redis_fps, queue_start=x['queued'], queue_end=y['queued'], queue_max=max(s['channels'][sid]['queued'] for s in samples), stale=y['stale'] - x['stale'], drop=y['drop'] - x['drop'], missed=y['observation_missed'] - x['observation_missed'], stage_fps={k:(y['counts'].get(k,0)-x['counts'].get(k,0))/dt for k in y['counts']}, redis_length=y['redis'].get('length'), redis_pttl=y['redis'].get('pttl'))) pm = {k:b['publisher'].get(k,0)-v for k,v in a['publisher'].items() if isinstance(v,(int,float)) and isinstance(b['publisher'].get(k),(int,float))} p.save(label + '-window.json', dict(samples=samples, complete=True, rows=rows, publisher_delta=pm)) print(label, [(v['stream_id'], round(v['result_fps'],2), None if v['redis_fps'] is None else round(v['redis_fps'],2), v['queue_end'], v['stale']) for v in rows], flush=True) return rows def stop_all(label): rt.PHASE = label assert all(s in rt.ORIGINAL or p.quiescent(v) for s,v in rt.states().items()) for sid in rt.ORIGINAL: q = rt.stop(sid) p.save(label + '-' + sid + '.json', dict(state=q['state'], run_id=q.get('run_id'), generation=q.get('run_generation'), cleanup=q.get('last_cleanup'))) def recreate(expected_image, label): p.ledger('recreate', phase=label, expected_image=expected_image) p.cmd(COMPOSE + ['config', '--quiet']) p.cmd(COMPOSE + ['up', '--no-start', '--no-deps', '--force-recreate', '--pull', 'never', 'analytics']) p.cmd(['docker', 'start', 'contextra-analytics-1']) deadline = time.monotonic() + 120 while time.monotonic() < deadline: current = p.base.inspect('contextra-analytics-1') if current['State'].get('Health', {}).get('Status') == 'healthy': break time.sleep(1) else: raise RuntimeError('READINESS_FAILED') assert current['Image'] == expected_image plan = p.read('plan.json') assert sorted(current['Config']['Env']) == sorted(plan['analytics']['Config']['Env']) assert current['Mounts'] == plan['analytics']['Mounts'], 'MOUNTS_CHANGED' for key in ('NanoCpus','CpuQuota','CpuPeriod','CpusetCpus','Memory','MemorySwap','RestartPolicy'): assert current['HostConfig'][key] == plan['analytics']['HostConfig'][key], key assert model_configs() == plan['configs'], 'MODEL_CONFIG_CHANGED' p.save(label + '-ready.json', dict(image=current['Image'], container=current['Id'], models_unchanged=True)) def start_all(label): rt.PHASE = label plan = p.read('plan.json') for src, sid in p.CHANNELS.items(): assert p.api('/api/v1/streams/' + sid + '/settings')['configured'] == plan['settings'][src]['configured'] rt.resource() q = rt.start(sid) assert q['inference_profile']['active_models'] == {'det': 'RFDETRMedium'} print(label, sid, 'Medium first Frame observed', flush=True) def run(): assert not (R / 'run-start.json').exists(), 'RUN_ALREADY_STARTED' plan = p.read('plan.json') assert time.time() < plan['deadline'] assert p.sha(p.ACTIVE) == plan['active_hash'] and p.sha(p.ENV) == plan['env_hash'] assert p.sha(p.prior.REG) == plan['registry_hash'] assert p.base.inspect('contextra-analytics-1')['Image'] == plan['analytics']['Image'] p.save('run-start.json', dict(at=time.time(), heads=plan['head'], image_only=True)) rows = window('baseline', 5) stop_all('before-candidate') active_before = (R / 'active.before').read_bytes() env_before = (R / 'env.before').read_bytes() active_changed = env_changed = None try: cfg = yaml.safe_load(active_before) cfg['services']['analytics']['image'] = IMAGE['image'] active_data = yaml.safe_dump(cfg).encode() lines = env_before.decode().splitlines(keepends=True) indices = [i for i,v in enumerate(lines) if v.startswith('CONTEXTRA_IMAGE=')] assert len(indices) == 1 lines[indices[0]] = 'CONTEXTRA_IMAGE=' + IMAGE['tag'] + '\n' env_data = ''.join(lines).encode() p.save('mutation-intent.json', dict(active_before=plan['active_hash'], env_before=plan['env_hash'], active_after=hashlib.sha256(active_data).hexdigest(), env_after=hashlib.sha256(env_data).hexdigest())) p.atomic(p.ACTIVE, active_data, plan['active_hash']) active_changed = p.sha(p.ACTIVE) p.atomic(p.ENV, env_data, plan['env_hash']) env_changed = p.sha(p.ENV) recreate(IMAGE['image'], 'candidate') start_all('candidate-start') time.sleep(5) rows += window('candidate', 20) except Exception as error: p.save('failure.json', dict(phase=rt.PHASE, type=type(error).__name__, reason=str(error)[:200])) print('TRIAL FAILED', rt.PHASE, type(error).__name__, str(error)[:100], flush=True) finally: stop_all('restore-stop') if active_changed: p.atomic(p.ACTIVE, active_before, active_changed) if env_changed: p.atomic(p.ENV, env_before, env_changed) recreate(plan['analytics']['Image'], 'restored') start_all('restored-start') time.sleep(5) rows += window('restored', 5) assert p.sha(p.ACTIVE) == plan['active_hash'] and p.sha(p.ENV) == plan['env_hash'] assert p.sha(p.prior.REG) == plan['registry_hash'] assert p.base.inspect('contextra-redis-1')['Id'] == plan['redis'] sender = p.prior.ops.prior.prev.sender() assert all(sender.get(k) == plan['sender'].get(k) for k in ('executionId', 'uri', 'video', 'mode', 'state')) p.save('final.json', dict(at=time.time(), rows=rows, image=plan['analytics']['Image'], settings_unchanged=True, active_persistent_restored=True, registry_unchanged=True, redis_preserved=True, sender_preserved=True, states=rt.states(), publisher=p.api('/api/v1/metadata/status'), resources=rt.resource())) with (R / 'summary.csv').open('w') as f: writer = csv.DictWriter(f, fieldnames=[k for k in rows[0] if k != 'stage_fps']) writer.writeheader() writer.writerows({k:v for k,v in row.items() if k != 'stage_fps'} for row in rows) print('RESTORED: original image/config/settings; Medium3 running; sender/Redis preserved', flush=True) def finalize(): plan = p.read('plan.json') current = p.base.inspect('contextra-analytics-1') assert current['Image'] == plan['analytics']['Image'] assert p.sha(p.ACTIVE) == plan['active_hash'] and p.sha(p.ENV) == plan['env_hash'] assert p.sha(p.prior.REG) == plan['registry_hash'] assert p.base.inspect('contextra-redis-1')['Id'] == plan['redis'] sender = p.prior.ops.prior.prev.sender() assert all(sender.get(k) == plan['sender'].get(k) for k in ('executionId', 'uri', 'video', 'mode', 'state')) current_states = rt.states() for src, sid in p.CHANNELS.items(): assert p.api('/api/v1/streams/' + sid + '/settings')['configured'] == plan['settings'][src]['configured'] assert current_states[sid]['state'] == 'running' rows = [] for label in ('baseline', 'candidate', 'restored'): window = p.read(label + '-window.json') assert window['complete'] rows += window['rows'] p.save('final.json', dict(at=time.time(), rows=rows, image=current['Image'], settings_unchanged=True, active_persistent_restored=True, registry_unchanged=True, redis_preserved=True, sender_preserved=True, states=current_states, publisher=p.api('/api/v1/metadata/status'), resources=rt.resource(), verification_note='Sender execution identity compared; live packet counters naturally change. Read-only finalization, no repeat runtime trial.')) with (R / 'summary.csv').open('w') as f: writer = csv.DictWriter(f, fieldnames=[k for k in rows[0] if k != 'stage_fps']) writer.writeheader() writer.writerows({k:v for k,v in row.items() if k != 'stage_fps'} for row in rows) print('FINAL VERIFIED: original image/settings/three runs; sender/Redis preserved', flush=True) if __name__ == '__main__': {'gate': gate, 'run': run, 'finalize': finalize}[sys.argv[1]]()