import copy
import csv
import hashlib
import json
import os
from pathlib import Path
import resource
import shutil
import subprocess
import sys
import time
import traceback

os.umask(0o077)
resource.setrlimit(resource.RLIMIT_CPU,(180,180))
sys.path.insert(0,'/tmp/codex/ltb1021/2026-09-16_contextra-channel-process_r01')
import process_ops as old
import yaml

R=Path(__file__).parent
R.chmod(0o700)
p,rt,REPO=old.p,old.rt,old.REPO
p.R=rt.R=R
sys.path.insert(0,str(REPO/'tools'))
from test_run_manager import load_preset,select_changes
MODEL=Path('/data/contextra/models/triton/RFDETRMedium')
CONFIG=MODEL/'config.pbtxt'
PRESETS=Path('/workspace/contextra-runtime-config/test-run/presets/medium16')
MANIFEST=Path('/workspace/contextra-runtime-config/test-run/medium16-registry.json')
ORIGINAL=list(rt.ORIGINAL)
PHASE='gate'


def api(path,method='GET',body=None):
    started=time.monotonic()
    if method!='GET':p.ledger('request',phase=PHASE,path=path,method=method)
    try:return p.api(path,method,body)
    except Exception as error:
        p.ledger('request_error',phase=PHASE,path=path,method=method,error=type(error).__name__,seconds=time.monotonic()-started)
        raise


def states():return {q['stream_id']:q for q in api('/api/v1/streams')['streams']}


def triton(suffix):
    code="import urllib.request;print(urllib.request.urlopen('http://127.0.0.1:8000/"+suffix+"',timeout=5).read().decode())"
    return json.loads(p.cmd(['docker','exec','contextra-analytics-1','python3','-c',code],timeout=10))


def actual_models():
    return {n:triton('v2/models/'+n+'/config') for n in ('RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke')}


def guard():return rt.resource()


def changes_for(desired,current):
    changes=select_changes(desired,current)
    for key in ('pose','fire'):
        if key in desired:
            value={k:v for k,v in desired[key].items() if current.get(key,{}).get(k)!=v}
            if value:changes[key]=value
            else:changes.pop(key,None)
    return changes


def gate():
    assert not (R/'plan.json').exists(),'GATE_EXISTS'
    assert p.cmd(['git','rev-parse','HEAD']).strip()=='d357942a19fdd5abf14617edf9a9a708a6d7fd98'
    assert not p.cmd(['git','status','--porcelain']).strip()
    image=json.loads((old.R/'image2.json').read_text())
    container=p.base.inspect('contextra-analytics-1')
    assert container['Image']==image['image']
    assert all(p.sha(REPO/n)==v for n,v in image['hashes'].items())
    env=dict(v.split('=',1) for v in container['Config']['Env'])
    assert env['CONTEXTRA_STREAM_EXECUTION_MODE']=='process'
    current=states();reg=yaml.safe_load(p.prior.REG.read_text());registered={q['stream_id']:q for q in reg['added']}
    assert all(s in ORIGINAL or p.quiescent(q) for s,q in current.items()),'OTHER_ACTIVE_CHANNEL'
    settings={s:api('/api/v1/streams/'+s+'/settings')['configured'] for s in ORIGINAL}
    for sid in ORIGINAL:
        q,cfg,src=current[sid],settings[sid],registered[sid]
        assert q['state']=='running' and q['inference_profile']['active_models']=={'det':'RFDETRMedium'}
        assert cfg['analysisFpsTarget']==10 and not cfg['tracking']['enabled'] and not cfg['intrusion']['enabled']
        assert not cfg['pose']['enabled'] and not cfg['fire']['enabled'] and not cfg['debug_all_candidates']
        assert not src['store'] and not src['active_learning'] and not src.get('capture',{}).get('enabled')
        assert not src.get('autostart',False) and not q['metadata_path']['performance']['detail']
    models=actual_models();medium=models['RFDETRMedium']
    assert medium['max_batch_size']==1 and len(medium['instance_group'])==1 and medium['instance_group'][0]['count']==1
    assert not medium.get('dynamic_batching')
    mount=next(m for m in container['Mounts'] if m['Destination']=='/opt/contextra/models/triton/RFDETRMedium')
    assert mount['Source']==str(MODEL) and not mount['RW']
    assert [int(v) for v in medium['version_policy']['specific']['versions']]==[2]
    paths=[p.ACTIVE,p.ENV,p.prior.REG,CONFIG]
    for path,name in zip(paths,['active.before','env.before','registry.before','medium-config.before']):shutil.copy2(path,R/name)
    channels=[]
    for group,sid in enumerate(ORIGINAL,1):
        channels.append(dict(group=group,copy=1,stream_id=sid,source_id=registered[sid]['source_id'],original=True,origin=sid))
    for copy_no in range(2,7):
        for group,count in enumerate((6,5,5),1):
            if copy_no<=count:
                channels.append(dict(group=group,copy=copy_no,stream_id=f'bench16-g{group}-c{copy_no}',
                    source_id=f'bench16-source-g{group}-c{copy_no}',original=False,origin=ORIGINAL[group-1]))
    assert len(channels)==16
    existing_manifest=json.loads(MANIFEST.read_text()) if MANIFEST.exists() else None
    if existing_manifest is not None:assert existing_manifest['channels']==channels,'PRESET_OWNER_CONFLICT'
    for row in channels[3:]:
        if row['stream_id'] in current:
            assert existing_manifest is not None and registered[row['stream_id']]['source_id']==row['source_id']
        assert not any(s['source_id']==row['source_id'] and s['stream_id']!=row['stream_id'] for s in registered.values() if s.get('source_id'))
    publisher=api('/api/v1/metadata/status')['metrics']
    assert publisher['workerMode']=='process-channel' and publisher['workerLimit']==16
    assert publisher['channels']+sum(r['stream_id'] not in current for r in channels[3:])<=64
    probe="""import json,urllib.request
from google.protobuf import text_format
from tritonclient.grpc import model_config_pb2
from pathlib import Path
v=model_config_pb2.ModelConfig();text_format.Parse(Path('/opt/contextra/models/triton/RFDETRMedium/config.pbtxt').read_text(),v)
assert not v.instance_group and v.max_batch_size==1 and not v.HasField('dynamic_batching')
g=v.instance_group.add();g.count=4;g.kind=model_config_pb2.ModelInstanceGroup.KIND_GPU;g.gpus.append(0)
metrics=urllib.request.urlopen('http://127.0.0.1:8002/metrics',timeout=5).read().decode()
lines=[x for x in metrics.splitlines() if 'RFDETRMedium' in x and not x.startswith('#')]
print(json.dumps(dict(config=text_format.MessageToString(v),metrics=lines,
 server=json.load(urllib.request.urlopen('http://127.0.0.1:8000/v2',timeout=5)),
 trace=json.load(urllib.request.urlopen('http://127.0.0.1:8000/v2/trace/setting',timeout=5)))))
"""
    candidate=json.loads(p.cmd(['docker','exec','contextra-analytics-1','python3','-c',probe],timeout=15))
    (R/'medium-config.candidate').write_text(candidate.pop('config'))
    p.save('triton-capabilities.json',candidate)
    p.save('plan.json',dict(at=time.time(),head=p.cmd(['git','rev-parse','HEAD']).strip(),analytics=container,
        image=image,models=models,engine_hash=p.sha(MODEL/'2/model.plan'),config_hash=p.sha(CONFIG),
        candidate_hash=p.sha(R/'medium-config.candidate'),registry=reg,states=current,settings=settings,channels=channels,
        existing_manifest=existing_manifest,persistent_hashes={str(x):p.sha(x) for x in (p.ACTIVE,p.ENV)},
        redis=p.base.inspect('contextra-redis-1')['Id'],inventory=p.base.inventory(),sender=p.prior.ops.prior.prev.sender(),
        resources=guard(),publisher=publisher,
        caps=dict(start_seconds=90,stop_seconds=90,readiness_seconds=120,helper_cpu_seconds=180,helper_ram_gib=1,
            analytics_ram_gib=16,host_available_gib=8,gpu_mib=40000,disk_free_gib=15,log_mib=30,
            warmup_seconds=5,stages={'4':20,'8':20,'12':20,'16':60}),
        restore='Stop added13 and retain registrations/presets; keep original3 running. Retain instance4 if accepted/stable; revert1 for instability/resource hazard.'))
    print('Gate PASS: process image; Medium2 instance1/batch1; registry '+str(len(current))+'; reusable13 prepared in ledger',flush=True)


def prepare(plan):
    global PHASE
    PHASE='prepare'
    PRESETS.mkdir(mode=0o700,parents=True,exist_ok=True)
    existing={s['stream_id']:s for s in plan['registry']['added']}
    allowed=set(api('/openapi.json')['components']['schemas']['StreamSpec']['properties'])
    for row in plan['channels'][3:]:
        sid=row['stream_id'];src=existing[row['origin']];cfg=plan['settings'][row['origin']]
        values={k:copy.deepcopy(cfg[k]) for k in ('display_score','display_categories','debug_all_candidates','intrusion','frame_time_anchor')}
        values.update(tracking={'enabled':False},analysisFpsTarget=10)
        preset=dict(version=1,camera=dict(stream_id=sid,source_id=row['source_id'],server_id=src['server_id'],uri=src['uri']),
            settings=values,debug_metrics=False,wait_seconds=90)
        path=PRESETS/(sid+'.yaml')
        if path.exists():assert yaml.safe_load(path.read_text())==preset,'PRESET_DRIFT'
        else:
            with path.open('x') as f:yaml.safe_dump(preset,f,allow_unicode=True)
        load_preset(path)
        if sid not in states():
            assert not (R/(sid+'-register-request.json')).exists(),'NO_REGISTER_RETRY'
            body={k:copy.deepcopy(v) for k,v in src.items() if k in allowed}
            body.update(preset['camera'],source_type='rtsp',store=False,active_learning=False,autostart=False,
                        scenarios='off',analysisFpsTarget=10)
            p.save(sid+'-register-request.json',dict(body=body,at=time.time()))
            response=api('/api/v1/streams?start=false','POST',body)
            assert response['state']=='stopped'
        current=api('/api/v1/streams/'+sid+'/settings')['configured']
        desired=dict(values,downstream_score=cfg['downstream_score'],pose={'enabled':False},fire={'enabled':False})
        changes=changes_for(desired,current)
        if changes:api('/api/v1/streams/'+sid+'/settings','PATCH',dict(expected_revision=current['revision'],**changes))
        after=api('/api/v1/streams/'+sid+'/settings')['configured']
        assert not changes_for(desired,after)
        assert after['execution_plan']==cfg['execution_plan'],'MODEL_PLAN_CHANGED'
        p.save(sid+'-prepared.json',dict(configured=after,preset=str(path)))
    registry=yaml.safe_load(p.prior.REG.read_text())
    regs={s['stream_id']:s for s in registry['added']}
    for row in plan['channels']:
        s=regs[row['stream_id']]
        assert not s.get('autostart',False) and not s.get('store') and not s.get('active_learning')
        assert not s.get('capture',{}).get('enabled')
    assert all(states()[r['stream_id']]['state']=='stopped' for r in plan['channels'][3:])
    manifest=dict(channels=plan['channels'],purpose='Reusable Medium16 benchmark, stopped retained; no sender control',
        source_commit=plan['head'],instance_candidate=4,max_batch=1,presets=str(PRESETS))
    if not MANIFEST.exists():
        with MANIFEST.open('x') as f:json.dump(manifest,f,indent=2)
    else:assert json.loads(MANIFEST.read_text())['channels']==plan['channels']
    p.save('all-prepared.json',dict(registry=registry,hash=p.sha(p.prior.REG),at=time.time()))
    print('Prepared13; all start=false/autostart=false, settings complete',flush=True)


def stop_one(sid,label):
    q=states()[sid]
    if p.quiescent(q):return q
    run=q['run_id'];generation=q['run_generation']
    marker=sid+'-stop-'+str(generation)+'-'+run+'.json'
    assert not (R/marker).exists(),'STOP_ALREADY_REQUESTED'
    p.save(marker,dict(at=time.time(),phase=label,run_id=run,generation=generation))
    try:response=api('/api/v1/streams/'+sid+'/stop','POST')
    except Exception as error:response=dict(error=type(error).__name__)
    deadline=time.monotonic()+90
    while time.monotonic()<deadline:
        q=states()[sid];cleanup=q.get('last_cleanup') or {}
        assert q['run_id']==run and q['run_generation']==generation
        if q.get('child_pid') is None and cleanup.get('generation')==generation and cleanup.get('child_reaped'):
            p.save(label+'-'+sid+'-cleanup.json',dict(state=q,response=response))
            assert cleanup.get('completed') and not cleanup.get('forced') and cleanup.get('writer_released'),'ABNORMAL_CHILD_CLEANUP'
            return q
        time.sleep(.5)
    p.save(label+'-'+sid+'-cleanup-timeout.json',q)
    raise RuntimeError('STOP_TIMEOUT_NO_RETRY')


def reflect(plan,count,label):
    global PHASE
    PHASE=label
    assert all(p.quiescent(q) for q in states().values()),'LIVE_CHANNEL_BEFORE_REFLECTION'
    p.ledger('analytics_restart',phase=label,instance=count)
    p.cmd(['docker','stop','--time','75','contextra-analytics-1'],timeout=85)
    p.cmd(['docker','start','contextra-analytics-1'],timeout=30)
    end=time.monotonic()+120
    while time.monotonic()<end:
        current=p.base.inspect('contextra-analytics-1')
        if current['State'].get('Health',{}).get('Status')=='healthy':break
        time.sleep(1)
    else:raise RuntimeError('READINESS_FAILED')
    assert current['Image']==plan['analytics']['Image'] and current['Id']==plan['analytics']['Id']
    assert current['Mounts']==plan['analytics']['Mounts'] and current['Config']['Env']==plan['analytics']['Config']['Env']
    actual=actual_models();expected=copy.deepcopy(plan['models'])
    assert sum(g['count']*len(g.get('gpus',[0])) for g in actual['RFDETRMedium']['instance_group'])==count
    expected['RFDETRMedium']['instance_group']=actual['RFDETRMedium']['instance_group']
    assert actual==expected,'MODEL_CONFIG_DRIFT'
    assert not actual['RFDETRMedium'].get('dynamic_batching')
    assert p.sha(MODEL/'2/model.plan')==plan['engine_hash']
    assert triton('v2/models/RFDETRMedium')['versions']==['2']
    p.save(label+'-ready.json',dict(at=time.time(),configs=actual,resources=guard()))
    print(label,'healthy; actual Medium instances',count,flush=True)


def start_one(sid,label):
    global PHASE
    PHASE=label;guard();q=states()[sid]
    if q['state']=='running':return q
    assert p.quiescent(q)
    marker=label+'-'+sid+'-start.json'
    assert not (R/marker).exists(),'START_ALREADY_REQUESTED'
    p.save(marker,dict(at=time.time()))
    api('/api/v1/streams/'+sid+'/start','POST')
    end=time.monotonic()+90
    while time.monotonic()<end:
        q=states()[sid]
        if q['state']=='running' and q.get('metadata_path',{}).get('processed',0)>0:
            assert q['inference_profile']['active_models']=={'det':'RFDETRMedium'}
            assert q.get('child_pid') and not q['scenarios']['active']
            p.save(label+'-'+sid+'-first-frame.json',dict(at=time.time(),state=q,resources=guard()))
            print('Started',sid,'PID',q['child_pid'],flush=True)
            return q
        if q['state']=='error':
            p.save(label+'-'+sid+'-failure.json',q)
            raise RuntimeError('START_FAILED')
        time.sleep(.5)
    p.save(label+'-'+sid+'-timeout.json',q)
    raise RuntimeError('FIRST_FRAME_TIMEOUT')


def triton_sample():
    code="""import json,urllib.request
stats=json.load(urllib.request.urlopen('http://127.0.0.1:8000/v2/models/RFDETRMedium/versions/2/stats',timeout=5))
raw=urllib.request.urlopen('http://127.0.0.1:8002/metrics',timeout=5).read().decode()
lines=[v for v in raw.splitlines() if 'RFDETRMedium' in v and not v.startswith('#')]
print(json.dumps(dict(stats=stats,metrics=lines)))
"""
    at=time.monotonic();value=json.loads(p.cmd(['docker','exec','contextra-analytics-1','python3','-c',code],timeout=12))
    value['at']=at;return value


def sample(rows):
    before=states();at=time.monotonic();channels={}
    for row in rows:
        sid=row['stream_id'];q=before[sid];path=q.get('metadata_path',{});perf=path.get('performance',{})
        channels[sid]=dict(state=q['state'],pid=q.get('child_pid'),run_id=q.get('run_id'),generation=q.get('run_generation'),
            error=q.get('last_error'),processed=path.get('processed',0),queue=path.get('queued',0),
            stale=path.get('stale',0),drop=path.get('queue_drop',0),counts=perf.get('counts',{}),
            observation_missed=perf.get('observation_missed',0),collector=path.get('collector_frames'),
            publisher=q.get('publisher_metrics',{}),redis=rt.redis_sample(row))
    return dict(at=at,channels=channels,publisher=api('/api/v1/metadata/status')['metrics'],triton=triton_sample(),resource=guard())


def summarize(stage,samples,complete):
    a,b=samples[0],samples[-1];dt=b['at']-a['at'];rows=[]
    for sid,x in a['channels'].items():
        y=b['channels'][sid];assert (x['run_id'],x['generation'])==(y['run_id'],y['generation'])
        rx,ry=x['redis'],y['redis']
        redis=None if rx.get('missing') or ry.get('missing') else (ry['entries']-rx['entries'])/(ry['at']-rx['at'])
        rows.append(dict(stage=stage,stream_id=sid,seconds=dt,pid=y['pid'],result_fps=(y['processed']-x['processed'])/dt,
            redis_fps=redis,**{k+'_fps':(y['counts'].get(k,0)-x['counts'].get(k,0))/dt for k in ('decoded','selected','infer','native','publisher_submit')},
            queue_start=x['queue'],queue_end=y['queue'],queue_max=max(s['channels'][sid]['queue'] for s in samples),
            stale=y['stale']-x['stale'],drop=y['drop']-x['drop'],missed=y['observation_missed']-x['observation_missed'],
            replaced=y['publisher'].get('replaced',0)-x['publisher'].get('replaced',0),
            send_error=y['publisher'].get('send_error',0)-x['publisher'].get('send_error',0),
            redis_length=ry.get('length'),redis_pttl=ry.get('pttl')))
    ta,tb=a['triton'],b['triton'];td=tb['at']-ta['at']
    x,y=ta['stats']['model_stats'][0],tb['stats']['model_stats'][0]
    totals={k:int(y[k])-int(x[k]) for k in ('inference_count','execution_count')}
    stages={}
    for k,last in y['inference_stats'].items():
        first=x['inference_stats'][k];cnt=int(last['count'])-int(first['count']);ns=int(last['ns'])-int(first['ns'])
        stages[k]=dict(count=cnt,total_ns=ns,mean_ms=ns/cnt/1e6 if cnt else None)
    pending=[]
    for sample_ in samples:
        for line in sample_['triton']['metrics']:
            if line.startswith('nv_inference_pending_request_count{'):pending.append(float(line.rsplit(' ',1)[1]))
    resources=[s['resource'] for s in samples]
    cpus=[(v['cpu']['usage_usec']-u['cpu']['usage_usec'])/1e6/(v['at']-u['at']) for u,v in zip(resources,resources[1:])]
    report=dict(stage=stage,complete=complete,rows=rows,publisher_delta={k:b['publisher'].get(k,0)-a['publisher'].get(k,0) for k in ('replaced','expired','send_error','byte_limit','message_limit')},
        workers=b['publisher']['workers'],triton=dict(seconds=td,**totals,inference_fps=totals['inference_count']/td,
            stats=stages,pending_max=max(pending) if pending else None,instance_distribution=None,
            instance_distribution_reason='Installed standard metrics/stats aggregate by model/version; no instance label. No invasive trace/backend modification.'),
        resources=dict(memory_gib=[min(v['memory'] for v in resources)/1024**3,max(v['memory'] for v in resources)/1024**3],
            cpu_cores=[min(cpus),max(cpus)],gpu_util=[min(v['gpu'][0][2] for v in resources),max(v['gpu'][0][2] for v in resources)],
            gpu_mib=[min(v['gpu'][0][0] for v in resources),max(v['gpu'][0][0] for v in resources)],
            available_ram_gib=min(v['available'] for v in resources)/1024**3,disk_free_gib=min(v['disk'] for v in resources)/1024**3))
    report['accepted']=complete and all(r['result_fps']>=9.5 and r['redis_fps'] is not None and r['redis_fps']>=9.5
        and r['stale']==0 and r['drop']==0 and r['queue_end']<=2 and r['send_error']==0 for r in rows)
    p.save('stage'+str(stage)+'-summary.json',report)
    with (R/('stage'+str(stage)+'.csv')).open('w') as f:
        w=csv.DictWriter(f,fieldnames=list(rows[0]));w.writeheader();w.writerows(rows)
    print('Stage',stage,'accepted',report['accepted'],'result',round(min(r['result_fps'] for r in rows),2),round(max(r['result_fps'] for r in rows),2),
          'queue',max(r['queue_end'] for r in rows),'stale',sum(r['stale'] for r in rows),flush=True)
    return report


def observe(rows,seconds):
    global PHASE
    stage=len(rows);PHASE='stage'+str(stage)
    time.sleep(5)
    samples=[sample(rows)];end=samples[0]['at']+seconds;bad=0
    try:
        while time.monotonic()<end:
            time.sleep(min(5,max(0,end-time.monotonic())))
            current=sample(rows);samples.append(current)
            p.save('stage'+str(stage)+'-window.json',dict(complete=False,samples=samples))
            assert len({q['pid'] for q in current['channels'].values()})==stage
            if any(q['state']!='running' or q['error'] for q in current['channels'].values()):raise RuntimeError('CHANNEL_RUNTIME_FAILURE')
            a,b=samples[-2],samples[-1];dt=b['at']-a['at']
            worsening=any((b['channels'][s]['stale']-q['stale']>=3 or b['channels'][s]['drop']-q['drop']>=3)
                or (b['channels'][s]['queue']>q['queue']+2 and b['channels'][s]['queue']>2) for s,q in a['channels'].items())
            bad=bad+1 if worsening else 0
            if bad>=2:raise RuntimeError('SUSTAINED_BACKLOG')
        report=summarize(stage,samples,True)
        p.save('stage'+str(stage)+'-window.json',dict(complete=True,samples=samples))
        return report
    except Exception:
        if len(samples)>1:summarize(stage,samples,False)
        raise


def final_check(plan,instance,failed):
    current=states();reg=yaml.safe_load(p.prior.REG.read_text());specs={v['stream_id']:v for v in reg['added']}
    originals={v['stream_id']:v for v in plan['registry']['added']}
    for sid,original in originals.items():assert specs[sid]==original,'ORIGINAL_REGISTRY_CHANGED'
    for sid in ORIGINAL:
        q=current[sid]
        assert q['state']=='running' and q['last_error'] is None and q['cleanup_error'] is None
        assert api('/api/v1/streams/'+sid+'/settings')['configured']==plan['settings'][sid]
        assert not q['metadata_path']['performance']['detail']
    for row in plan['channels'][3:]:
        sid=row['stream_id']
        if sid not in current:continue
        assert p.quiescent(current[sid]) and current[sid].get('child_pid') is None
        assert not specs[sid].get('autostart',False)
    cfg=actual_models()
    assert cfg['RFDETRMedium']['instance_group'][0]['count']==instance
    assert p.sha(CONFIG)==(plan['candidate_hash'] if instance==4 else plan['config_hash'])
    assert p.sha(MODEL/'2/model.plan')==plan['engine_hash']
    assert all(p.sha(Path(path))==sha for path,sha in plan['persistent_hashes'].items())
    assert p.base.inspect('contextra-redis-1')['Id']==plan['redis']
    sender=p.prior.ops.prior.prev.sender()
    assert all(sender[k]==plan['sender'][k] for k in ('executionId','uri','video','mode','state'))
    p.save('final.json',dict(at=time.time(),instance=instance,failed=failed,states=current,
        publisher=api('/api/v1/metadata/status'),configs=cfg,resources=guard(),original_settings_preserved=True,
        sender_redis_preserved=True,engine_image_preserved=True,temporary_registrations_retained=True,
        trace_enabled=False,diagnostics_off=True))
    print('FINAL instance',instance,'original3 running; added registered/stopped',flush=True)


def run(resume=False):
    global PHASE
    plan=p.read('plan.json')
    if resume:
        assert p.read('failure.json')['phase']=='prepare' and p.read('final.json')['instance']==1
        assert not (R/'preparation-resume.json').exists(),'CORRECTION_ALREADY_USED'
        assert all(states()[s]['run_id']==plan['states'][s]['run_id'] for s in ORIGINAL)
        shutil.copy2(R/'failure.json',R/'preparation-failure.json')
        shutil.copy2(R/'final.json',R/'preparation-final.json')
        p.save('preparation-resume.json',dict(at=time.time(),reason='Compare partial Pose/Fire fields, not the entire normalized settings object. No inference or model mutation occurred.'))
    else:
        assert not (R/'run-start.json').exists(),'RUN_ALREADY_STARTED'
        assert p.sha(p.prior.REG)==p.sha(R/'registry.before')
        p.save('run-start.json',dict(at=time.time()))
    assert p.sha(CONFIG)==plan['config_hash']
    failed=False;changed=False;reflected=False
    try:
        prepare(plan)
        for sid in ORIGINAL:stop_one(sid,'before-instance4')
        p.ledger('config_apply',before=plan['config_hash'],after=plan['candidate_hash'])
        p.atomic(CONFIG,(R/'medium-config.candidate').read_bytes(),plan['config_hash']);changed=True
        reflect(plan,4,'instance4');reflected=True
        for n,seconds in ((4,20),(8,20),(12,20),(16,60)):
            for row in plan['channels'][:n]:start_one(row['stream_id'],'ascent')
            observe(plan['channels'][:n],seconds)
    except Exception as error:
        failed=True
        p.save('failure.json',dict(at=time.time(),phase=PHASE,error=type(error).__name__,reason=str(error)[:400],trace=traceback.format_exc()))
        print('STOP ASCENT',PHASE,type(error).__name__,str(error)[:150],flush=True)
        p.cmd(['docker','logs','--since','3m','--tail','180','contextra-analytics-1'],timeout=15)
    finally:
        PHASE='cleanup'
        for row in reversed(plan['channels'][3:]):
            if row['stream_id'] in states():stop_one(row['stream_id'],'added-stop')
        if failed and changed:
            for sid in ORIGINAL:stop_one(sid,'restore-stop')
            p.atomic(CONFIG,(R/'medium-config.before').read_bytes(),plan['candidate_hash'])
            reflect(plan,1,'restore-instance1')
            instance=1
        else:instance=4 if reflected else 1
        for sid in ORIGINAL:start_one(sid,'final-original')
        final_check(plan,instance,failed)


if __name__=='__main__':
    if len(sys.argv)>1:run(resume=sys.argv[1]=='resume-preparation')
    else:gate()