from prepare import g
import datetime,json,os,pathlib,re,subprocess,tempfile,time,urllib.request,urllib.parse
from model_tooling.deploy import deployment_plan
from model_tooling.medium_acceptance import POLICY
R,REPO,ENV,CFG=g.R,g.REPO,g.ENV,g.CFG
os.umask(0o077)
def save():
    fd,name=tempfile.mkstemp(dir=R)
    with os.fdopen(fd,'w') as f:json.dump(result,f,indent=2);f.flush();os.fsync(f.fileno())
    os.replace(name,R/'runtime.json')
def inspect():return next(x for x in g.inspect_all() if x['Name']=='/contextra-analytics-1')
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
def api(path,method='GET',timeout=5):
    with opener.open(urllib.request.Request(base+path,method=method),timeout=timeout) as f:return json.load(f)
def gpu():return int(g.shell(['nvidia-smi','--id=0','--query-gpu=memory.used','--format=csv,noheader,nounits']).strip())
def threads():
    code='''import pathlib,json
c={"processes":0,"tasks":0,"media":0}
for p in pathlib.Path('/proc').glob('[0-9]*'):
 try:
  args=(p/'cmdline').read_bytes().split(bytes([0]))
  if len(args)<3 or args[1:3]!=[b'-m',b'app.main']:continue
  c['processes']+=1
  for q in (p/'task').iterdir():
   name=(q/'comm').read_text().lower();c['tasks']+=1
   if any(k in name for k in ['rtp','rtsp','infer','preprocess','appsink','nvvideoconvert']):c['media']+=1
 except (FileNotFoundError,ProcessLookupError):pass
print(json.dumps(c))'''
    return json.loads(g.shell(['docker','exec','contextra-analytics-1','python3','-c',code]))
def safety_stop():
    if inspect()['State']['Status']=='running':
        subprocess.run(['docker','stop','--timeout','20','contextra-analytics-1'],check=True,capture_output=True,timeout=35)
    result['analytics_safety_stop']=True;save()
def states(s):return {k:s.get(k) for k in ['state','desired_state','observed_state','cleanup_pending','cleanup_error','last_error','run_generation','per_task_buffers']}
def observation():
    result['observation_elapsed_s']=round(time.monotonic()-started,3);save()
    try:
        p=subprocess.run(['python3',str(R/'observe.py'),base],capture_output=True,text=True,timeout=12)
        result['external_observation']=json.loads(p.stdout) if p.returncode==0 else {'status':'HELPER_FAILED'}
    except Exception as e:result['external_observation']={'status':type(e).__name__}
    # timer is armed at bootstrap, not at timeout; give it at most 12s from Start.
    until=started+12
    while not (R/'diagnostics/python-stack.txt').exists() or (R/'diagnostics/python-stack.txt').stat().st_size==0:
        if time.monotonic()>=until:break
        time.sleep(.2)
    markers=[]
    for p in (R/'diagnostics').glob('*.jsonl'):
        try:
            for line in p.read_text().splitlines():
                row=json.loads(line)
                markers.append({k:row.get(k) for k in ['stage','phase','thread_role','monotonic_ns','diagnostic_error']})
        except Exception:result['marker_read_failed']=True
    result['markers']=sorted(markers,key=lambda x:x['monotonic_ns'])[:100]
    stack=R/'diagnostics/python-stack.txt'
    result['stack_bytes']=stack.stat().st_size if stack.exists() else 0
    result['stack_elapsed_s']=round(time.monotonic()-started,3)
    save()

assert not (R/'runtime.json').exists()
gate=json.loads((R/'gate.json').read_text());build=json.loads((R/'build.json').read_text())
assert json.loads((R/'image-verify.json').read_text())['status']=='PASS'
assert g.shell(['git','-C',str(REPO),'rev-parse','HEAD']).strip()==g.HEAD
g.assert_source()
assert g.sha(ENV)==gate['env_hash'] and g.sha(CFG)==gate['config_hash']
assert g.sha(CFG.parent/'2/model.plan')==gate['plan_hash']
c=inspect();assert c['Image']==gate['old_image'] and c['State']['Status']=='exited'
env=g.load_env_file(str(ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
assert g.sha(overlay)==gate['overlay_hash']
assert g.other_baseline(g.inspect_all())==gate['other_containers']
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'
stream=g.yaml.safe_load(overlay.read_text())['added'][0]
path='/api/v1/streams/'+urllib.parse.quote(stream['stream_id'],safe='')
base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
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(ENV),'-f',str(REPO/'docker-compose.yml'),'-f',str(R/'diagnostics-compose.yaml')]
result=dict(stage='GATE_PASS',start_attempts=0,stop_attempts=0,source=g.HEAD,image=build['image_id'],version1_rollback=False)
save();changed=False;started=None
try:
    lines=ENV.read_text().splitlines(keepends=True);matches=[i for i,l in enumerate(lines) if l.split('=',1)[0].strip()=='CONTEXTRA_IMAGE']
    assert len(matches)==1;lines[matches[0]]='CONTEXTRA_IMAGE='+build['tag']+'\n'
    assert g.sha(ENV)==gate['env_hash']
    fd,name=tempfile.mkstemp(dir=ENV.parent)
    with os.fdopen(fd,'w') as f:os.fchmod(f.fileno(),0o600);f.write(''.join(lines));f.flush();os.fsync(f.fileno())
    os.replace(name,ENV);changed=True
    result.update(stage='ENV_APPLIED',env_after_hash=g.sha(ENV));save()
    g.validate_deployment_inputs(str(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))
    assert render['services']['analytics']['image']==build['tag']
    assert render['services']['analytics']['environment']['CONTEXTRA_START_DIAGNOSTICS_DIR']==diag
    with (R/'compose.log').open('w') as log:
        subprocess.run(compose+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],check=True,stdout=log,stderr=subprocess.STDOUT,timeout=120)
    end=time.monotonic()+300;ready=False
    while time.monotonic()<end:
        c=inspect();assert c['Image']==build['image_id'] and not c['RestartCount'] and not c['State']['OOMKilled']
        assert c['HostConfig']['RestartPolicy']['Name']=='no'
        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(3)
    assert ready
    versions=json.loads(g.shell(['docker','exec','contextra-analytics-1','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']
    streams=api('/api/v1/streams')['streams'];assert len(streams)==1 and streams[0]['state']=='stopped'
    s=api(path);assert s['state']=='stopped' and s['run_generation']==0 and s['run_id'] is None
    # Verify the real app UID against the mounted owner without creating a fake marker.
    code='import os,stat; p=os.environ["CONTEXTRA_START_DIAGNOSTICS_DIR"];s=os.stat(p);assert s.st_uid==os.geteuid() and stat.S_IMODE(s.st_mode)==0o700;assert os.access(p,os.W_OK);print("OK")'
    assert g.shell(['docker','exec','contextra-analytics-1','python3','-c',code]).strip()=='OK'
    result.update(stage='READY',model_versions=versions,initial_stopped=True,diagnostics_writable=True,threads_before=threads());save()
    initial_gpu=gpu();result['gpu_baseline_mib']=initial_gpu
    since=datetime.datetime.now(datetime.timezone.utc).isoformat();started=time.monotonic();first=None;success=False
    try:
        assert g.sha(overlay)==gate['overlay_hash'] and g.sha(CFG)==gate['config_hash']
        result.update(stage='WINDOW_STARTED',start_attempts=1);save();api(path+'/start','POST',10)
        while time.monotonic()-started<60:
            s=api(path,timeout=3);result['last_state']=states(s)
            result['roles']={role:{k:data.get(k) for k in ['state','source_stage','stage_counts','stage_first_at']} for role,data in s.get('roles',{}).items()}
            frames=(s.get('per_task_buffers') or {}).get('det',0);result['frames']=frames
            used=gpu();result['gpu_peak_mib']=max(result.get('gpu_peak_mib',initial_gpu),used)
            if used-initial_gpu>8192:result['budget_exceeded']=True;break
            if s.get('last_error'):break
            if frames and s['observed_state']=='running':
                if first is None:first=time.monotonic()
                if frames>=100 or time.monotonic()-first>=10:success=True;break
            time.sleep(.5)
    except Exception as e:
        result['window_exception']=type(e).__name__;save()
    finally:
        observation()
        result['stop_attempts']=1;save()
        try:
            api(path+'/stop','POST',25)
            post=api(path);result['post_stop']=states(post)
            result['cleanup_verified']=all(post[k]=='stopped' for k in ['state','desired_state','observed_state']) and not post.get('cleanup_pending') and not post.get('cleanup_error')
            result['threads_after']=threads()
            result['worker_cleanup_verified']=result['cleanup_verified'] and result['threads_after']['media']==0
            if not result['worker_cleanup_verified']:safety_stop()
        except Exception as e:
            result.update(cleanup_verified=False,cleanup_exception=type(e).__name__);safety_stop()
    result['window_seconds']=round(time.monotonic()-started,3)
    log=subprocess.run(['docker','logs','--since',since,'--tail','250','contextra-analytics-1'],capture_output=True,text=True,timeout=15)
    result['geometry_samples']=[]
    for line in (log.stdout+log.stderr).splitlines():
        prefix='[NATIVE_BRIDGE_SAMPLE] '
        if prefix in line and len(result['geometry_samples'])<3:
            result['geometry_samples'].append(json.loads(line.split(prefix,1)[1]))
    result['stage']='PIPELINE_OBSERVED' if success else 'PIPELINE_INCOMPLETE'
except Exception as e:
    result.update(stage='FAILED',exception_type=type(e).__name__)
    if changed:safety_stop()
finally:
    result['other_containers_unchanged']=g.other_baseline(g.inspect_all())==gate['other_containers']
    result['config_unchanged']=g.sha(CFG)==gate['config_hash'];result['overlay_unchanged']=g.sha(overlay)==gate['overlay_hash']
    c=inspect()
    if c['State']['Status']!='running' or result.get('worker_cleanup_verified'):
        try:
            subprocess.run(['docker','update','--restart=on-failure:3','contextra-analytics-1'],check=True,capture_output=True,timeout=15)
            c=inspect();result['restart_restored']=c['HostConfig']['RestartPolicy']==gate['original_restart']
            override['services']['analytics']['restart']='on-failure:3'
            (R/'diagnostics-compose.yaml').write_text(g.yaml.safe_dump(override))
        except Exception as e:result['restart_restore_error']=type(e).__name__
    result['analytics']={k:c['State'].get(k) for k in ['Status','OOMKilled','ExitCode']}
    result['analytics']['restart']=c['RestartCount'];result['diagnostic_override_retained']=True
    save();print(json.dumps({k:v for k,v in result.items() if k not in ['markers','geometry_samples','roles']},indent=2))