"""Existing bounded live harness; one approved Pose file window, no model conversion."""
import json
from pathlib import Path

R=Path(__file__).parent
model=json.loads((R/'model-prepared.json').read_text())
source=Path('/tmp/codex/ltb1021/2026-09-13_contextra-intrusion-live_r01/runtime.py').read_text()
source=source.replace("cfg['services']['analytics']['image']=built['image_id']", """cfg['services']['analytics']['image']=built['image_id']
        model=json.loads((R/'model-prepared.json').read_text())
        assert g.sha(Path(model['directory'])/'config.pbtxt')==model['config_hash']
        assert g.sha(Path(model['directory'])/'2/model.plan')==model['plan_sha256']
        target='/opt/contextra/models/triton/RFDETRKeypointPreview'
        volumes=cfg['services']['analytics'].setdefault('volumes',[])
        assert not any(v.get('target')==target for v in volumes if isinstance(v,dict))
        volumes.append({'type':'bind','source':model['directory'],'target':target,'read_only':True,'bind':{'create_host_path':False}})""")
source=source.replace("assert versions==['2']", """assert versions==['2']
        pose_versions=json.loads(r.cmd(['docker','exec',g.NAME,'python3','-c',
            'import json,urllib.request;print(json.dumps(json.load(urllib.request.urlopen("http://localhost:8000/v2/models/RFDETRKeypointPreview",timeout=3))["versions"]))']).stdout)
        assert pose_versions==['2']
        state['pose_version']=pose_versions""")
source=source.replace("'debug_all_candidates':False, 'intrusion':", """'debug_all_candidates':False,
            'pose':{'enabled':True,'show_bbox':True,'show_joints':True,'show_edges':True,
                    'person_score':.25,'joint_score':.15,'min_joints':10,
                    'capture_interval_s':3.,'nms_enabled':True,'nms_iou_threshold':.85}, 'intrusion':""")
source=source.replace("assert configured['display_categories']==[] and configured['intrusion']['enabled']", """assert configured['display_categories']==[] and configured['intrusion']['enabled']
        assert configured['execution_plan']['model']=='RFDETRKeypointPreview'
        assert configured['execution_plan']['role']=='pose'""")
source=source.replace("end=time.monotonic()+2\n        while time.monotonic()<end:\n            r.sample()\n            readback_frames()\n            time.sleep(.25)", """end=time.monotonic()+40
        warmed=False
        while time.monotonic()<end:
            sample=r.sample()
            last=sample['metadata_path']['performance'].get('last_frame',{}).get('result')
            readback_frames()
            if last and last[1]>=20*10**9:
                warmed=True
                break
            time.sleep(.2)
        assert warmed,'WARMUP_PTS_TIMEOUT'""")
source=source.replace('end=time.monotonic()+8','end=time.monotonic()+15')
source=source.replace("assert value['frameTime']['rtpTimestamp'] is None", """assert value['frameTime']['rtpTimestamp'] is None
        assert value['keypointSchema']['names'][11]=='left_hip'
        assert value['nms']['input']-value['nms']['kept']==value['nms']['suppressed']
        assert all(len(o['keypoints'])==17 and o['modelKey']=='RFDETRKeypointPreview' for o in value['objects'])""")
source=source.replace("state['result_events_created']=after['metadata_path'].get('events_created',0)", """state['result_events_created']=after['metadata_path'].get('events_created',0)
        state['last_metadata_path']=after['metadata_path']
        state['pose_only']=list(after['roles'])==['pose']
        assert state['pose_only']
        state['nms_live']={k:v for k,v in after['metadata_path'].items() if k.startswith('nms_')}""")
source=source.replace("state['error_code']=str(e) if", """import traceback
        (R/'runtime-error.private.txt').write_text(traceback.format_exc())
        state['error_code']=str(e) if""")
exec(compile(source,'bounded_pose_runtime','exec'))