"""One offline service window. Reuse bounded HTTP/log/resource helpers, not old main."""
import importlib.util
import json
import os
from pathlib import Path
import signal
import time

from build import R, g, HEAD

os.umask(0o077)
loader = importlib.util.spec_from_file_location('bounded_runtime',
    '/tmp/codex/ltb1021/2026-09-12_contextra-snapshot-tracking_r01/stage2/runtime-derived.py')
r = importlib.util.module_from_spec(loader)
loader.loader.exec_module(r)
a, plan, built = r.a, r.plan, r.built
state = r.state
state.update(stage='prepared', runtime_windows=0, caps=dict(helper_seconds=600, readiness_seconds=180,
    first_result_seconds=60, measurement_seconds=8, cleanup_seconds=60, frames=3000,
    gpu_delta_mib=8192, host_ram_floor_gib=4, images=3, capture_seconds=60))
active = r.C/'compose.yaml'
flag = R/'visual/capture.enabled'
normal = ['docker','compose','--env-file',str(g.ENV),'-f',str(g.REPO/'docker-compose.yml'),'-f',str(active)]
after_overlay = after_active = after_env = None
log = None
started = reflected = False
events = {}
matches = set()


def save():
    r.save()


def readback_frames():
    entries = r.redis_query('reader','XREVRANGE',r.stream,'+','-','COUNT','150') or []
    values = {}
    for _, fields in entries:
        if isinstance(fields, list):
            fields = dict(zip(fields[::2], fields[1::2]))
        value = json.loads(fields['json'])
        if value['runId'] != state['run_id']:
            continue
        assert all(o['classId'] == 0 and o['confidence'] > .5 for o in value['objects'])
        assert value['frameTime']['rtpTimestamp'] is None
        assert value['frameTime']['detectedAtWallMs'] is None
        values[value['frameSeq']] = value
        for event in value.get('events', []):
            assert event['deliveryStatus'] == 'NOT_SUBMITTED' and event['clipUri'] is None and event['thumbnailUri'] is None
            if len(events) < 3:
                events[event['eventId']] = event
    for path in (R/'visual').glob('capture-*/frame-*.json'):
        own = json.loads(path.read_text())
        wire = values.get(own['frameSeq'])
        if wire is not None:
            assert wire == own, 'FRAME_MISMATCH'
            matches.add(own['frameSeq'])
            output = R/'samples'/path.name
            if not output.exists():
                output.write_text(json.dumps(wire, indent=2))
    state['redis_visual_matches'] = sorted(matches)
    state['events_observed'] = len(events)
    state['redis_frame_count_in_sample'] = len(values)
    if events:
        (R/'event-samples.json').write_text(json.dumps(list(events.values()), indent=2))
    save()


def ready(deadline):
    while time.monotonic() < deadline:
        c = g.inspect()
        assert c['Image'] == built['image_id'] and c['State']['Status'] == 'running'
        assert not c['State']['OOMKilled'] and not c['RestartCount']
        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 main():
    global after_overlay, after_active, after_env, log, started, reflected
    assert not (R/'run.json').exists()
    v=json.loads((R/'image-verify.json').read_text())
    assert v['errors'] == v['failures'] == 0
    source=json.loads((R/'source.json').read_text())
    assert all(g.sha(g.REPO/p)==h for p,h in source['hashes'].items())
    assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()==HEAD
    input_data=json.loads((R/'approved-input.json').read_text())
    assert g.sha(Path(input_data['path']))==input_data['hash']
    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'] and r.api(r.path)[1]['state']=='stopped'
    assert g.others()==plan['other_containers']
    assert r.redis_query('reader','PING')=='PONG'
    save()
    signal.signal(signal.SIGALRM, r.alarm)
    signal.alarm(600)
    try:
        cfg=g.yaml.safe_load(active.read_text())
        cfg['services']['analytics']['image']=built['image_id']
        for key in ('CONTEXTRA_METADATA_VISUAL_DIR','CONTEXTRA_START_DIAGNOSTICS_DIR','CONTEXTRA_TEST_FRAME_ANCHOR_FILE'):
            cfg['services']['analytics'].setdefault('environment',{})[key]=''
        after_active=a.atomic(active,g.yaml.safe_dump(cfg).encode(),plan['active_hash'])
        state['active_after']=after_active
        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'])
        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)
        after_overlay=a.atomic(a.overlay,g.yaml.safe_dump(data).encode(),plan['overlay_hash'])
        state.update(env_after=after_env,overlay_after=after_overlay)
        diagnostic={'services':{'analytics':{'restart':'no','environment':{
            'CONTEXTRA_METADATA_VISUAL_DIR':'/run/contextra-visual',
            'CONTEXTRA_START_DIAGNOSTICS_DIR':'/run/contextra-boundary'},'volumes':[
            {'type':'bind','source':input_data['path'],'target':'/run/contextra-offline/input.mp4','read_only':True,'bind':{'create_host_path':False}},
            {'type':'bind','source':str(R/'visual'),'target':'/run/contextra-visual','bind':{'create_host_path':False}},
            {'type':'bind','source':str(R/'diagnostics'),'target':'/run/contextra-boundary','bind':{'create_host_path':False}}]}}}
        (R/'diagnostic-compose.yaml').write_text(g.yaml.safe_dump(diagnostic))
        compose=normal+['-f',str(R/'diagnostic-compose.yaml')]
        state['compose']=compose
        save()
        r.cmd(compose+['config','--quiet'])
        reflected=True
        state['analytics_reflected']=True
        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(time.monotonic()+180)
        assert r.api(r.path)[1]['state']=='stopped'
        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/RFDETRMedium",timeout=3))["versions"]))']).stdout)
        assert versions==['2']
        old=r.api(r.path+'/settings')[1]
        desired={'expected_revision':old['configured']['revision'], 'display_categories':[],
            'debug_all_candidates':False, 'intrusion':{'enabled':True, 'roi_name':'Entry A','debug_roi':True,
            'roi':[[.2,.2],[.8,.2],[.8,.95],[.2,.95]]}}
        configured=r.api(r.path+'/settings','PATCH',desired)[1]['configured']
        assert configured['display_categories']==[] and configured['intrusion']['enabled']
        after_overlay=g.sha(a.overlay)
        state['overlay_after']=after_overlay
        state['trial_revision']=configured['revision']
        state['readiness']='PASS'
        save()
        started=True
        state['runtime_windows']=1
        r.command('start')
        end=time.monotonic()+60
        first=None
        while time.monotonic()<end:
            item=r.sample()
            assert log.error is None
            assert not item.get('last_error') and not item.get('cleanup_error'), 'PIPELINE_ERROR'
            if item['state']=='running' and item.get('metadata_path',{}).get('processed',0):
                first=item
                break
            time.sleep(.25)
        assert first is not None, 'FIRST_RESULT_TIMEOUT'
        state['run_id']=first['run_id']
        state['first_result_ns']=time.monotonic_ns()
        generation=first['run_generation']
        r.api(r.path+'/metrics','POST',dict(run_generation=generation,enabled=True,seconds=60))
        end=time.monotonic()+2
        while time.monotonic()<end:
            r.sample()
            readback_frames()
            time.sleep(.25)
        # Capture after confirmation so the bounded images can show active warnings.
        flag.touch(mode=0o600,exist_ok=False)
        before=r.sample()
        begin=time.monotonic_ns()
        end=time.monotonic()+8
        while time.monotonic()<end:
            item=r.sample()
            assert item['state']=='running' and not item.get('cleanup_error'), 'PIPELINE_ERROR'
            readback_frames()
            assert r.api('/live')[0]==200
            time.sleep(.25)
        after=r.sample()
        state['measurement']=dict(start_ns=begin,end_ns=time.monotonic_ns(),
            before=before['metadata_path']['performance'],after=after['metadata_path']['performance'])
        r.api(r.path+'/metrics','POST',dict(run_generation=generation,enabled=False,seconds=60))
        readback_frames()
        state['result_events_created']=after['metadata_path'].get('events_created',0)
        state['window_result']='PASS' if matches and events else 'OBSERVATION_WAIT'
        save()
    except Exception as e:
        state['error']=type(e).__name__
        state['error_code']=str(e) if str(e) in ('PIPELINE_ERROR','FIRST_RESULT_TIMEOUT','FRAME_MISMATCH','HELPER_CAP') else 'BOUNDED_STEP_FAILED'
        if reflected:
            state['failed_state']=g.inspect()['State']
        save()
    finally:
        signal.alarm(0)
        if started:
            try:
                r.command('stop')
                end=time.monotonic()+60
                while time.monotonic()<end:
                    item=r.sample()
                    if item.get('cleanup_error') or item.get('last_error'):
                        break
                    if item['state']=='stopped' and not item.get('cleanup_pending'):
                        state['cleanup_verified']=bool((item.get('last_cleanup') or {}).get('completed'))
                        break
                    time.sleep(.25)
            except Exception as e:
                state['stop_error']=type(e).__name__
        flag.unlink(missing_ok=True)
        success=state.get('cleanup_verified') and not state.get('error') and not state.get('stop_error')
        try:
            if log:
                log.close()
                state['log_bytes']=log.bytes
                state['log_error']=log.error
            if reflected:
                state['pre_restore_state']=g.inspect()['State']
            if after_overlay:
                a.atomic(a.overlay,(R/'overlay.before').read_bytes(),after_overlay)
                state['input_settings_restored']=True
            if reflected and success:
                # Required to remove temporary file/capture mounts and reload the original stopped input.
                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(time.monotonic()+180)
                assert r.api(r.path)[1]['state']=='stopped'
                actual=g.inspect()
                assert actual['HostConfig']['RestartPolicy']==plan['analytics']['HostConfig']['RestartPolicy']
                assert not any(m['Destination'] in ('/run/contextra-offline/input.mp4','/run/contextra-visual','/run/contextra-boundary') for m in actual['Mounts'])
                state['temporary_mounts_removed']=True
                state['final']='HEALTHY_STOPPED_NEW_IMAGE'
            elif reflected:
                r.cmd(['docker','stop','--timeout','20',g.NAME],40)
                r.cmd(['docker','update','--restart=no',g.NAME])
                cfg=g.yaml.safe_load(active.read_text())
                cfg['services']['analytics']['restart']='no'
                state['active_after']=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']
            assert g.sha(g.CFG)==plan['config_hash'] and g.sha(g.CFG.parent/'2/model.plan')==plan['plan_hash']
            state['non_targets_model_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'
        save()
        print(json.dumps({k:state.get(k) for k in ('window_result','cleanup_verified','redis_visual_matches',
            'events_observed','error','error_code','restore_error','final','temporary_mounts_removed')}))


if __name__=='__main__':
    main()