import json, os, pathlib, re, signal, subprocess, tempfile, time, urllib.parse, urllib.request
from prepare import g, R, HEAD, source_cas
from tools.validate_deployment import validate_deployment_inputs
os.umask(0o077)
gate=json.loads((R/'gate.json').read_text());build=json.loads((R/'build.json').read_text())
assert not (R/'runtime.json').exists()
assert json.loads((R/'image-verify.json').read_text())['status']=='PASS'
source_cas()
env=g.load_env_file(str(g.ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
spec=g.yaml.safe_load(overlay.read_text())['added'][0]
path='/api/v1/streams/'+urllib.parse.quote(spec['stream_id'],safe='')
base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
# URI is read in memory only; output contains counters, never endpoints or process IDs.
uri=next((value for key,value in spec.items() if isinstance(value,str) and value.startswith('rtsp://')),None)
peer=urllib.parse.urlsplit(uri) if uri else None
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
result=dict(stage='prepared',start_attempts=0,stop_attempts=0,readiness_budget_s=300,
            first_frame_budget_s=60,after_first_s=10,pre_stop_frame_cap=100,total_probe_buffer_cap=1000,
            stop_observation_s=60,gpu_increment_cap_mib=8192,host_ram_floor_gib=4,
            total_helper_budget_s=600,post_stop=[])
changed=False;cid=None;env_after=None
def save():
    p=R/'runtime.partial';p.write_text(json.dumps(result,indent=2));p.replace(R/'runtime.json')
def api(p,method='GET',timeout=5):
    with opener.open(urllib.request.Request(base+p,method=method),timeout=timeout) as f:return json.load(f)
def state(v):
    d={k:v.get(k) for k in ['state','desired_state','observed_state','cleanup_pending','run_generation','per_task_buffers','runtime_observation']}
    for name in ['last_error','cleanup_error']:
        value=v.get(name)
        d[name]={k:value.get(k) for k in ['error_code','failure_stage','role']} if isinstance(value,dict) else None
    return d
def common_cas():
    assert g.sha(g.CFG)==gate['config_hash'] and g.sha(overlay)==gate['overlay_hash']
    assert g.sha(g.CFG.parent/'2/model.plan')==gate['plan_hash']
def current():
    c=g.inspect()
    if cid is not None:assert c['Id']==cid and c['Image']==build['image_id']
    return c
def safety_stop():
    c=current()
    if c['State']['Status']=='running':
        subprocess.run(['docker','stop','--timeout','20',g.NAME],check=True,capture_output=True,timeout=35)
    result['analytics_safety_stop']=True;save()
def resources():
    used=g.gpu()
    mem={l.split(':')[0]:int(l.split()[1]) for l in pathlib.Path('/proc/meminfo').read_text().splitlines()}
    assert used-gate['gpu_mib']<8192 and mem['MemAvailable']>4*1024**2, 'RESOURCE_CAP'
    return {'gpu_mib':used,'ram_available_mib':mem['MemAvailable']//1024}
def receive():
    if not peer or not peer.hostname:return {'status':'PEER_UNAVAILABLE'}
    try:
        init=current()['State']['Pid']
        text=subprocess.check_output(['nsenter','--target',str(init),'--net','ss','-H','-t','-n','-i','-p'],text=True,stderr=subprocess.DEVNULL,timeout=3)
        entries=[];entry=None
        for line in text.splitlines():
            if line and not line[0].isspace():
                if entry is not None:entries.append(entry)
                entry=line
            elif entry is not None:entry+=' '+line.strip()
        if entry is not None:entries.append(entry)
        values=[];unknown=0
        for entry in entries:
            fields=entry.split()
            if len(fields)<5 or fields[4]!=peer.hostname+':'+str(peer.port or 554):continue
            pids=re.findall(r'pid=(\d+)',entry)
            owned=False
            for pid in pids:
                try:
                    cmd=(pathlib.Path('/proc')/pid/'cmdline').read_bytes().split(bytes([0]))
                    owned=owned or any(cmd[i:i+2]==[b'-m',b'app.main'] for i in range(len(cmd)-1))
                except OSError:pass
            if not owned:continue
            amount=re.search(r'\bbytes_received:(\d+)',entry)
            if amount:values.append(int(amount.group(1)))
            else:unknown+=1
        return {'status':'OK','app_peer_connections':len(values)+unknown,'tcp_bytes_received_sum':sum(values) if not unknown else None,
                'unmetered_connections':unknown,'metric':'TCP_BYTES_NOT_RTP_COUNT'}
    except Exception as error:return {'status':type(error).__name__}
def observe():
    v=api(path,timeout=3)
    obs=v.get('runtime_observation') or []
    for row in obs:
        assert row['generation']==v['run_generation']
        assert max((x['count'] for x in row['counters'].values()),default=0)<=1000,'BUFFER_CAP'
    return {'monotonic_ns':time.monotonic_ns(),'state':state(v),'receive':receive(),'resources':resources()}
def alarm(*unused):raise TimeoutError('Overall deadline')
signal.signal(signal.SIGALRM,alarm);signal.alarm(600)
save()
try:
    common_cas();assert g.sha(g.ENV)==gate['env_hash']
    c=g.inspect();assert c['Id']==gate['container_id'] and c['Image']==gate['old_image'] and c['State']['Status']=='exited'
    assert c['HostConfig']['RestartPolicy']==gate['original_restart'] and g.others()==gate['other_containers']
    from model_tooling.deploy import deployment_plan
    from model_tooling.medium_acceptance import POLICY
    plan=deployment_plan(pathlib.Path('/data/contextra/model-artifacts/builds/medium-193-20260909-v4'),R/'unapplied-target',1,source_acceptance=POLICY)
    assert plan['source_acceptance']['status']=='LIMITED_USER_VISUAL_ACCEPTANCE'
    diag='/run/contextra-start-diagnostics'
    override={'services':{'analytics':{'restart':'no','environment':{'CONTEXTRA_START_DIAGNOSTICS_DIR':diag},
              'volumes':[{'type':'bind','source':str(R/'diagnostics'),'target':diag,'bind':{'create_host_path':False}}]}}}
    (R/'diagnostics-compose.yaml').write_text(g.yaml.safe_dump(override))
    compose=['docker','compose','--env-file',str(g.ENV),'-f',str(g.REPO/'docker-compose.yml'),'-f',str(R/'diagnostics-compose.yaml')]
    # gate.json preserves only the old image value and hashes, not credential-bearing env contents.
    lines=g.ENV.read_text().splitlines(keepends=True)
    indices=[i for i,line in enumerate(lines) if line.split('=',1)[0].strip()=='CONTEXTRA_IMAGE'];assert len(indices)==1
    lines[indices[0]]='CONTEXTRA_IMAGE='+build['tag']+'\n'
    assert g.sha(g.ENV)==gate['env_hash']
    fd,name=tempfile.mkstemp(dir=g.ENV.parent)
    with os.fdopen(fd,'w') as f:
        os.fchmod(f.fileno(),0o600);f.writelines(lines);f.flush();os.fsync(f.fileno())
    os.replace(name,g.ENV);changed=True;env_after=g.sha(g.ENV)
    result.update(stage='env_applied',env_after_hash=env_after);save()
    validate_deployment_inputs(str(g.ENV))
    subprocess.run(compose+['config','--quiet'],check=True,capture_output=True,timeout=30)
    render=json.loads(subprocess.check_output(compose+['config','--format','json'],text=True,timeout=30))['services']['analytics']
    assert render['image']==build['tag'] and render['restart']=='no' and render['environment']['CONTEXTRA_START_DIAGNOSTICS_DIR']==diag
    common_cas();assert g.sha(g.ENV)==env_after
    subprocess.run(compose+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],check=True,capture_output=True,timeout=120)
    cid=g.inspect()['Id'];result.update(stage='readiness',container_recreated=True);save()
    end=time.monotonic()+300;ready=False
    while time.monotonic()<end:
        c=current();resources()
        assert c['State']['Status']=='running' and not c['State']['OOMKilled'] and c['RestartCount']==0
        assert c['HostConfig']['RestartPolicy']=={'Name':'no','MaximumRetryCount':0}
        try:
            for endpoint in ['/live','/ready','/health']:api(endpoint)
            ready=c['State'].get('Health',{}).get('Status')=='healthy'
            if ready:break
        except Exception:pass
        time.sleep(2)
    assert ready
    versions=json.loads(g.sh(['docker','exec',g.NAME,'python3','-c',
        'import urllib.request,json;d=json.load(urllib.request.urlopen("http://localhost:8000/v2/models/RFDETRMedium",timeout=5));print(json.dumps(d["versions"]))']))
    assert versions==['2']
    listed=api('/api/v1/streams')['streams'];assert len(listed)==1 and listed[0]['state']=='stopped'
    initial=api(path);assert initial['state']=='stopped' and initial['run_generation']==0 and initial.get('run_id') is None
    result.update(stage='ready',model_versions=versions,initial_receive=receive(),resources_ready=resources());save()
    common_cas();assert g.sha(g.ENV)==env_after
    started=time.monotonic();first=None;result.update(stage='live',start_attempts=1);save()
    try:
        api(path+'/start','POST',10)
        while time.monotonic()-started<60:
            sample=observe();result['before_stop']=sample
            v=sample['state'];frames=(v.get('per_task_buffers') or {}).get('det',0)
            if v.get('last_error'):break
            if frames and v.get('observed_state')=='running':
                if first is None:first=time.monotonic();result['first_observed_s']=round(first-started,3)
                if frames>=100 or time.monotonic()-first>=10:break
            save();time.sleep(.5)
    except Exception as error:result['live_exception']=type(error).__name__
    finally:
        stopped=time.monotonic();result.update(stop_attempts=1,stop_requested_ns=time.monotonic_ns(),first_frame_confirmed=first is not None);save()
        try:
            response=api(path+'/stop','POST',25)
            result['stop_response']=state(response);result['stop_return_elapsed_s']=round(time.monotonic()-stopped,3);save()
            # Sticky error is deliberately ignored for sampling, not cleared or treated as cleanup success.
            while True:
                sample=observe();sample['after_stop_s']=round(time.monotonic()-stopped,3)
                result['post_stop'].append(sample);save()
                if time.monotonic()-stopped>=60:break
                time.sleep(min(3,max(0,60-(time.monotonic()-stopped))))
            final=result['post_stop'][-1]['state']
            result['cleanup_verified']=all(final.get(k)=='stopped' for k in ['state','desired_state','observed_state']) and not final.get('cleanup_pending') and not final.get('cleanup_error')
        except Exception as error:result['observation_exception']=type(error).__name__;result['cleanup_verified']=False
        if not result.get('cleanup_verified'):safety_stop()
    result['stage']='completed'
except Exception as error:
    result.update(stage='failed',exception_type=type(error).__name__);save()
    if changed:safety_stop()
finally:
    signal.alarm(0)
    if changed:
        try:
            c=current()
            if c['State']['Status']=='running' and not result.get('cleanup_verified'):safety_stop()
            assert c['HostConfig']['RestartPolicy']=={'Name':'no','MaximumRetryCount':0}
            subprocess.run(['docker','update','--restart=on-failure:3',g.NAME],check=True,capture_output=True,timeout=15)
            result['restart_restored']=current()['HostConfig']['RestartPolicy']==gate['original_restart']
        except Exception as error:result['restore_error']=type(error).__name__
    result['other_containers_unchanged']=g.others()==gate['other_containers']
    result['model_config_overlay_unchanged']=g.sha(g.CFG)==gate['config_hash'] and g.sha(overlay)==gate['overlay_hash'] and g.sha(g.CFG.parent/'2/model.plan')==gate['plan_hash']
    c=current();result['analytics']={k:c['State'].get(k) for k in ['Status','OOMKilled','ExitCode']};result['analytics']['restart_count']=c['RestartCount']
    result['gpu_final_mib']=g.gpu();result['diagnostic_override_retained']=changed
    result['env_cas_preserved']=env_after is None or g.sha(g.ENV)==env_after
    save()
    print(json.dumps({k:result.get(k) for k in ['stage','start_attempts','stop_attempts','first_frame_confirmed','stop_return_elapsed_s','cleanup_verified','analytics','restart_restored','other_containers_unchanged','model_config_overlay_unchanged','exception_type','observation_exception']},indent=2))