"""CPU-only comparison runner. Reuses approved SSH and resource guards."""
import concurrent.futures
import hashlib
import json
import pathlib
import subprocess
import sys

LOCAL=pathlib.Path(__file__).resolve().parent.parent
SOURCE=LOCAL.parent/'2026-10-28_contextra-pose-input-review-15clips_r01'
sys.path.insert(0,str(SOURCE/'helpers'))
import control as old

REMOTE=old.RB+'/'+LOCAL.name
PREVIOUS=old.REMOTE+'/download'
PREFIX='contextra-fight-ab-1028-r01-'
REPORT=LOCAL/'reports'
KEYS=('01_camera24_fight','07_KISA_fight_C00_015_0009','08_KISA_fight_C00_018_0012',
      '06_KISA_fight_C00_008_0005','04_camera29_fire','02_camera22_crowd')

def save(name,value):
    (REPORT/name).write_text(json.dumps(value,ensure_ascii=False,indent=2))

def snapshot(label):
    raw=old.checked(['docker','exec','-i','contextra-ai2-analytics-1','python3','-'],
        data=(old.PREP/'helpers/snapshot.py').read_bytes(),timeout=90)
    state=json.loads(raw)
    save('production_'+label+'.json',state)
    containers=old.checked(['docker','ps','--format','{{.Names}} {{.Image}} {{.ID}}']).decode()
    (REPORT/('containers_'+label+'.txt')).write_text(containers)
    assert state['health']['ready']
    return state

def setup():
    old.checked(['test','!','-e',REMOTE])
    save('resources_before.json',old.resources())
    assert old.checked(['docker','image','inspect',old.IMAGE,'--format','{{.Id}}']).decode().strip()==old.IMAGE
    specs=json.loads((SOURCE/'specs.json').read_text())
    selected={k:specs[k] for k in KEYS}
    download=SOURCE/'Contextra_Pose_Input_Review_20261028'
    hashes=json.loads((download/'SHA256SUMS.json').read_text())
    for k,n in zip(KEYS,(201,550,650,650,101,201)):
        for filename in ('cache_validation.json','observation_states.jsonl','frame_map.csv','identity_palette.json','input_rgb_sha256.json'):
            rel=k+'/data/'+filename
            assert hashlib.sha256((download/rel).read_bytes()).hexdigest()==hashes[rel]
        v=json.loads((download/k/'data/video_validation.json').read_text())
        assert v['frames']==n and v['nominal_fps']==10
    (LOCAL/'specs.json').write_text(json.dumps(selected,ensure_ascii=False,indent=2))
    baseline={}
    for name in ('fight_config.py','fight_runtime.py','fight_scenario.py'):
        raw=old.checked(['docker','exec','contextra-ai2-analytics-1','cat','/opt/contextra/app/'+name])
        (LOCAL/'baseline'/name).write_bytes(raw)
        current=pathlib.Path('/workspace/contextra-analytics-runtime-lifecycle-refactor/app')/name
        baseline[name]=dict(installed_sha256=hashlib.sha256(raw).hexdigest(),local_sha256=hashlib.sha256(current.read_bytes()).hexdigest(),same=raw==current.read_bytes())
    save('baseline_sources.json',baseline)
    old.checked(['mkdir',REMOTE],sudo=True)
    old.checked(['chown','si:si',REMOTE],sudo=True)
    old.checked(['chmod','700',REMOTE])
    old.checked(['mkdir','-p',REMOTE+'/download/validation']+[REMOTE+'/clips/'+k for k in KEYS])
    sync()
    snapshot('before')
    code="import os,json;print(json.dumps({k:os.environ.get(k,'<unset>') for k in ('CONTEXTRA_FIGHT_CURRENT_FRAME_FALLBACK','CONTEXTRA_FIGHT_EVENT_COOLDOWN_SECONDS')}))"
    save('baseline_private_flags.json',json.loads(old.checked(['docker','exec','contextra-ai2-analytics-1','python3','-c',code])))
    print('PREPARED cache=6 inference=0',flush=True)

def sync():
    for name in ('helpers','specs.json'):
        old.scp(str(LOCAL/name),'si@58.233.207.31:'+REMOTE+'/')

def trial(stage,clip):
    old.resources()
    spec=json.loads((LOCAL/'specs.json').read_text())[clip]
    name=PREFIX+stage+'-'+clip[:2]
    cmd=['docker','run','--rm','--pull','never','--name',name,'--user','0:0',
         '--read-only','--network','none','--cap-drop','ALL','--cap-add','DAC_OVERRIDE',
         '--security-opt','no-new-privileges','--cpus','2','--memory','8g','--pids-limit','256',
         '--shm-size','256m','--tmpfs','/tmp:rw,size=512m','--runtime','runc',
         '-e','NVIDIA_VISIBLE_DEVICES=void','-e','PYTHONDONTWRITEBYTECODE=1',
         '-e','OMP_NUM_THREADS=2','-e','OPENBLAS_NUM_THREADS=2','-e','CLIP='+clip,
         '-v',REMOTE+'/helpers:/helpers:ro','-v',REMOTE+'/specs.json:/specs.json:ro',
         '-v',REMOTE+'/clips/'+clip+':/eval','-v',spec['cache']+':/cache:ro',
         '-v',PREVIOUS+':/previous:ro','-v',REMOTE+'/download:/download',
         '-v',old.RB+'/'+old.SHORT.name+'/fonts/NotoSansCJK-Regular.ttc:/font.ttc:ro',
         '--entrypoint','python3',old.IMAGE,'/helpers/'+stage+'.py']
    print('START',stage,clip,flush=True)
    samples=[]
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        future=pool.submit(old.run,cmd,timeout=1800)
        try:
            while True:
                try:
                    p=future.result(timeout=30)
                    break
                except concurrent.futures.TimeoutError:
                    if future.done():raise
                    samples.append(old.resources())
                    print('RESOURCE',stage,clip,len(samples),flush=True)
        except BaseException:
            old.run(['docker','stop','--time','15',name],timeout=40)
            raise
    (REPORT/(stage+'_'+clip+'.log')).write_bytes(p.stdout+p.stderr)
    save(stage+'_'+clip+'_resources.json',samples)
    if p.returncode:raise RuntimeError(p.stderr.decode()[-1500:]+p.stdout.decode()[-1000:])
    print('DONE',stage,clip,p.stdout.decode()[-1200:],flush=True)

def finish_snapshot():
    after=snapshot('after');before=json.loads((REPORT/'production_before.json').read_text())
    stable={k:before[k]==after[k] for k in ('config_sha256','pose_engine_sha256','pose_config_sha256')}
    stream_fields=('state','run_id','run_generation','child_pid','cleanup_pending','last_error')
    changes={k:{f:[before['streams'].get(k,{}).get(f),v.get(f)] for f in stream_fields
        if before['streams'].get(k,{}).get(f)!=v.get(f)} for k,v in after['streams'].items()}
    changes={k:v for k,v in changes.items() if v}
    containers_same=(REPORT/'containers_before.txt').read_text()==(REPORT/'containers_after.txt').read_text()
    active=old.checked(['docker','ps','-a','--filter','name='+PREFIX,'--format','{{.Names}}']).decode().strip()
    assert not active
    from collections import Counter
    result=dict(config_model_unchanged=stable,containers_unchanged=containers_same,
        before_states=dict(Counter(s['state'] for s in before['streams'].values())),
        after_states=dict(Counter(s['state'] for s in after['streams'].values())),stream_changes=changes,
        trial_containers_remaining=[],production_control_calls=0,gpu_jobs=0)
    save('operational_comparison.json',result);save('resources_after.json',old.resources())
    print(json.dumps(result),flush=True)

if __name__=='__main__':
    if sys.argv[1]=='setup':setup()
    elif sys.argv[1]=='sync':sync()
    elif sys.argv[1]=='render':
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            for f in concurrent.futures.as_completed([pool.submit(trial,'render',clip) for clip in KEYS]):f.result()
    elif sys.argv[1] in ('evaluate','audit'):
        for clip in KEYS:trial(sys.argv[1],clip)
    elif sys.argv[1]=='test':trial('test_engines',KEYS[0])
    elif sys.argv[1]=='finish_snapshot':finish_snapshot()
    else:raise ValueError('Unsupported stage')