import importlib.util,json,signal,time
from pathlib import Path
from build import R,g,HEAD
s=importlib.util.spec_from_file_location('bounded','/tmp/codex/ltb1021/2026-09-12_contextra-snapshot-tracking_r01/stage2/runtime-derived.py')
r=importlib.util.module_from_spec(s);s.loader.exec_module(r)
a,plan,built=r.a,r.plan,r.built
state=r.state
state.update(stage='prepared',caps=dict(helper_s=600,readiness_s=120,first_frame_s=60,video_s=5.034,stop_s=60,frames=3000,gpu_delta_mib=8192,host_ram_floor_gib=4,log_bytes=2097152),runtime_windows=0)
active=r.C/'compose.yaml'
normal=['docker','compose','--env-file',str(g.ENV),'-f',str(g.REPO/'docker-compose.yml'),'-f',str(active)]
values={}
def ready():
    end=time.monotonic()+120
    while time.monotonic()<end:
        c=g.inspect();assert c['Image']==built['image_id'] and c['State']['Status']=='running' and not c['RestartCount']
        r.resource()
        try:
            if c['State'].get('Health',{}).get('Status')=='healthy' and r.api('/ready')[0]==200:return
        except Exception:pass
        time.sleep(1)
    raise TimeoutError('READINESS_TIMEOUT')
def frames():
    entries=r.redis_query('reader','XREVRANGE',r.stream,'+','-','COUNT','100') or []
    for _,fields in entries:
        if isinstance(fields,list):fields=dict(zip(fields[::2],fields[1::2]))
        v=json.loads(fields['json'])
        if v['runId']!=state.get('run_id'):continue
        assert v['frameTime']['rtpTimestamp'] is None and v['frameTime']['detectedAtWallMs'] is None
        assert not v.get('events')
        for o in v['objects']:
            assert o['classId'] in (0,1) and o['confidence']>=.25
            assert all(0<=x<=1 for x in o['bbox'])
        values[v['frameSeq']]=v
    state['redis_frames']=len(values)
    state['classes_seen']=sorted({o['classId'] for v in values.values() for o in v['objects']})
    matches=[]
    for p in (R/'samples').glob('sample-*.json'):
        raw=json.loads(p.read_text());wire=values.get(raw['frameSeq'])
        if wire is not None:
            raw['objects']=[o for o in raw['objects'] if o['visible']]
            assert raw==wire,'FRAME_MISMATCH'
            matches.append(raw['frameSeq'])
    state['same_frame_matches']=sorted(matches)
    r.save()
def main():
    assert not (R/'run.json').exists()
    assert json.loads((R/'image-gate.json').read_text())['runtime_gate']=='PASS'
    assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()==HEAD
    assert g.sha(active)==plan['active_hash'] and g.sha(g.ENV)==plan['env_hash'] and g.sha(a.overlay)==plan['overlay_hash']
    assert g.inspect()['Id']==plan['analytics']['Id']
    if g.inspect()['State']['Status']=='running':assert r.api(r.path)[1]['state']=='stopped'
    else:assert g.inspect()['State']['Status']=='exited' and g.inspect()['HostConfig']['RestartPolicy']['Name']=='no'
    assert g.others()==plan['other_containers']
    assert r.redis_query('reader','PING')=='PONG'
    inp=json.loads((R/'approved-input.json').read_text());model=json.loads((R/'model-prepared.json').read_text())
    assert g.sha(Path(inp['path']))==inp['sha256']
    assert g.sha(Path(model['directory'])/'2/model.plan')==plan['fire_plan_hash']
    state.update(apply_order=['image/env/model mount','offline input','recreate Analytics with restart=no','settings CAS','Start1','Stop1','restore overlay','recreate normal stopped'],restore='CAS original overlay; retain image and Fire mount on success; safe-stop on cleanup failure')
    r.save();signal.signal(signal.SIGALRM,r.alarm);signal.alarm(600)
    after_active=after_env=after_overlay=None;reflected=False;started=False;log=None
    try:
        cfg=g.yaml.safe_load(active.read_text());svc=cfg['services']['analytics'];svc['image']=built['image_id']
        target='/opt/contextra/models/triton/RFDETRFireSmoke'
        assert not any((v.get('target') if isinstance(v,dict) else v.split(':')[1])==target for v in svc.get('volumes',[]))
        svc.setdefault('volumes',[]).append(dict(type='bind',source=model['directory'],target=target,read_only=True,bind=dict(create_host_path=False)))
        for k in ('CONTEXTRA_METADATA_VISUAL_DIR','CONTEXTRA_START_DIAGNOSTICS_DIR','CONTEXTRA_TEST_FRAME_ANCHOR_FILE'):svc.setdefault('environment',{})[k]=''
        after_active=a.atomic(active,g.yaml.safe_dump(cfg).encode(),plan['active_hash']);state['active_after']=after_active;r.save()
        lines=g.ENV.read_text().splitlines(keepends=True);idx=[i for i,l in enumerate(lines) if l.split('=',1)[0]=='CONTEXTRA_IMAGE'];assert len(idx)==1
        lines[idx[0]]='CONTEXTRA_IMAGE='+built['tag']+'\n';after_env=a.atomic(g.ENV,''.join(lines).encode(),plan['env_hash']);state['env_after']=after_env;r.save()
        data=g.yaml.safe_load(a.overlay.read_text());test=data['added'][0]
        test.update(source_type='file',uri='/run/contextra-offline/input.mp4',frame_time_anchor=None);test.pop('source_options',None)
        assert test.get('store') is False and test.get('active_learning') is False and test.get('scenarios')=='off'
        after_overlay=a.atomic(a.overlay,g.yaml.safe_dump(data).encode(),plan['overlay_hash']);state['overlay_after']=after_overlay;r.save()
        (R/'samples/capture.enabled').touch(mode=0o600,exist_ok=False)
        diagnostic={'services':{'analytics':{'restart':'no','environment':{'CONTEXTRA_START_DIAGNOSTICS_DIR':'/run/contextra-boundary','CONTEXTRA_METADATA_SAMPLES_DIR':'/run/contextra-samples'},'volumes':[
            dict(type='bind',source=inp['path'],target='/run/contextra-offline/input.mp4',read_only=True,bind=dict(create_host_path=False)),
            dict(type='bind',source=str(R/'diagnostics'),target='/run/contextra-boundary',bind=dict(create_host_path=False)),
            dict(type='bind',source=str(R/'samples'),target='/run/contextra-samples',bind=dict(create_host_path=False))]}}}
        (R/'diagnostic-compose.yaml').write_text(g.yaml.safe_dump(diagnostic));compose=normal+['-f',str(R/'diagnostic-compose.yaml')]
        r.cmd(compose+['config','--quiet']);state['compose']=compose;state['analytics_reflected']=True;reflected=True;r.save()
        r.cmd(compose+['up','--no-start','--no-deps','--force-recreate','--pull','never','analytics'],120)
        r.cmd(['docker','start',g.NAME]);log=r.LogCapture(['docker','logs','--follow',g.NAME],R/'runtime.private.log',480)
        ready();assert r.api(r.path)[1]['state']=='stopped'
        probe='import json,urllib.request; print(json.dumps({m:json.load(urllib.request.urlopen("http://localhost:8000/v2/models/"+m,timeout=3))["versions"] for m in ("RFDETRFireSmoke","RFDETRMedium","RFDETRKeypointPreview")}))'
        versions=json.loads(r.cmd(['docker','exec',g.NAME,'python3','-c',probe]).stdout);assert all(v==['2'] for v in versions.values())
        state['model_versions']=versions;state['readiness']='PASS';r.save()
        old=r.api(r.path+'/settings')[1]['configured']
        desired=dict(expected_revision=old['revision'],fire=dict(enabled=True),pose=dict(enabled=False,skeleton=False),intrusion=dict(enabled=False),display_categories=['smoke_visible','flame_visible'],display_score=.25,analysisFpsTarget=10,debug_all_candidates=False)
        # Pose's current payload is retained except explicit enable and skeleton request.
        desired['pose']={'enabled':False,'show_joints':False,'show_edges':False}
        configured=r.api(r.path+'/settings','PATCH',desired)[1]['configured']
        after_overlay=g.sha(a.overlay);state['overlay_after']=after_overlay;state['trial_settings']=configured;r.save()
        assert configured['execution_plan']['model']=='RFDETRFireSmoke'
        state['runtime_windows']=1;started=True;r.command('start')
        end=time.monotonic()+60;first=None;last=None
        while time.monotonic()<end:
            v=r.sample();assert log.error is None
            assert not v.get('last_error') and not v.get('cleanup_error'),'PIPELINE_ERROR'
            if v.get('run_id'):state['run_id']=v['run_id']
            if v.get('metadata_path',{}).get('processed',0):
                if first is None:first=v;state['first_result_ns']=time.monotonic_ns()
                last=v;frames()
            if first is not None and (v['state']=='stopped' or time.monotonic_ns()-state['first_result_ns']>8e9):break
            time.sleep(.2)
        assert first is not None,'FIRST_RESULT_TIMEOUT'
        state['observed_eof_before_stop']=bool(last and last['state']=='stopped')
        state['last_live']=last;frames()
        assert values and state['classes_seen']==[0,1] and state['same_frame_matches'],'FIRE_REDIS_MAPPING_WAIT'
        for i,seq in enumerate(sorted(values)[:3]):(R/'samples'/f'frame-{i+1}.json').write_text(json.dumps(values[seq],indent=2))
        state['window_result']='PASS';r.save()
    except Exception as e:
        import traceback
        (R/'runtime-error.private.log').write_text(traceback.format_exc())
        state['error']=type(e).__name__;state['error_code']=str(e) if str(e) in ('PIPELINE_ERROR','FIRST_RESULT_TIMEOUT','FIRE_REDIS_MAPPING_WAIT') else 'BOUNDED_STEP_FAILED'
        if reflected:state['failed_state']=g.inspect()['State']
        r.save()
    finally:
        signal.alarm(0)
        (R/'samples/capture.enabled').unlink(missing_ok=True)
        if started:
            try:
                r.command('stop');end=time.monotonic()+60
                while time.monotonic()<end:
                    v=r.sample()
                    if v.get('cleanup_error') or v.get('last_error'):break
                    if v['state']=='stopped' and not v.get('cleanup_pending'):
                        state['cleanup_verified']=bool((v.get('last_cleanup') or {}).get('completed'));state['stop_status']=v;break
                    time.sleep(.25)
            except Exception as e:state['stop_error']=type(e).__name__
        if log:log.close();state['log_bytes']=log.bytes;state['log_error']=log.error
        success=state.get('window_result')=='PASS' and state.get('cleanup_verified') and not state.get('error') and not state.get('stop_error')
        try:
            if after_overlay:a.atomic(a.overlay,(R/'overlay.before').read_bytes(),after_overlay);state['input_settings_restored']=True
            if reflected and success:
                assert g.sha(active)==after_active and g.sha(g.ENV)==after_env
                r.cmd(normal+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],120);ready()
                assert r.api(r.path)[1]['state']=='stopped'
                actual=g.inspect();assert actual['HostConfig']['RestartPolicy']==plan['analytics']['HostConfig']['RestartPolicy']
                assert not any(m['Destination'].startswith('/run/contextra-') for m in actual['Mounts'])
                state['final']='HEALTHY_STOPPED_NEW_IMAGE';state['temporary_mounts_removed']=True
            elif reflected:
                r.cmd(['docker','update','--restart=no',g.NAME]);r.cmd(['docker','stop','--timeout','20',g.NAME],40)
                cfg=g.yaml.safe_load(active.read_text());cfg['services']['analytics']['restart']='no';a.atomic(active,g.yaml.safe_dump(cfg).encode(),after_active)
                state['final']='SAFE_STOPPED_NEW_IMAGE_RESTART_NO'
            else:
                if after_active:a.atomic(active,(R/'active-compose.before').read_bytes(),after_active)
                if after_env:a.atomic(g.ENV,(R/'env.before').read_bytes(),after_env)
                state['final']='UNCHANGED_SERVICE'
            assert g.others()==plan['other_containers'] and g.sha(g.CFG)==plan['config_hash'] and g.sha(g.CFG.parent/'2/model.plan')==plan['plan_hash']
            assert all(g.sha(Path(p))==h for p,h in plan['pose_hashes'].items())
            state['non_targets_unchanged']=True
        except Exception as e:
            state['restore_error']=type(e).__name__
            if reflected:
                try:r.cmd(['docker','update','--restart=no',g.NAME]);r.cmd(['docker','stop','--timeout','20',g.NAME],40);state['safety_stopped']=True
                except Exception:state['safety_stop_failed']=True
        state['stage']='completed';r.save()
        print(json.dumps({k:state.get(k) for k in ('window_result','cleanup_verified','redis_frames','classes_seen','error','error_code','final','restore_error')}))
if __name__=='__main__':main()