import csv
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tarfile
import time
import traceback

os.umask(0o077)
sys.path.insert(0, '/tmp/codex/ltb1021/2026-09-16_contextra-medium-fresh_r01')
import check as previous
import yaml

R = Path(__file__).parent
R.chmod(0o700)
p, rt = previous.p, previous.rt
p.R = rt.R = previous.R = R
REPO = rt.REPO
os.chdir(REPO)
FILES = ['app/analysis_metrics.py', 'app/multi_pipeline.py', 'app/det_result_path.py',
         'tests/test_pose_batch_path.py', 'tests/test_collector_metrics.py']


def gate():
    p.gate()
    plan = p.read('plan.json')
    assert plan['head'] == '19c6353bf96f6caba9a0d377bdecf0ad7a0a7368'
    assert all(sid in rt.ORIGINAL or p.quiescent(q) for sid,q in rt.states().items())
    for src,sid in p.CHANNELS.items():
        cfg = plan['settings'][src]['configured']
        assert not any(cfg[k]['enabled'] for k in ('pose','fire','tracking','intrusion'))
        assert cfg['analysisFpsTarget'] == 10
    plan['configs'] = previous.model_configs()
    assert plan['configs']['RFDETRMedium']['max_batch_size'] == 1
    assert plan['configs']['RFDETRMedium']['instance_group'][0]['count'] == 1
    plan['caps'] = dict(warmup=5, initial=20, post_saturation=30, max_after_first_frames=240,
        analytics_ram_gib=16, min_host_available_gib=8, gpu_mib=40000, disk_gib=15,
        helpers=1, receivers=0, profile_seconds_max=60, profile_samples_per_stage=256,
        build_seconds=600, prep_seconds=1200, total_seconds=2100)
    plan['deadline'] = datetime.fromisoformat('2026-09-16T14:27:25+09:00').timestamp() + 2100
    plan['restore'] = 'Success: candidate diagnostic image and Medium3 running; failure: exact pre-task image/env/Compose and original settings/running restored.'
    p.save('plan.json', plan)
    rt.resource()
    print('Gate PASS; Medium3, models/settings preserved', flush=True)


def build():
    assert p.read('saturation-source2.json')['success']
    assert not (R / 'image-start.json').exists()
    hashes = {name:p.sha(REPO / name) for name in FILES}
    digest = hashlib.sha256(json.dumps(hashes,sort_keys=True).encode()).hexdigest()[:8]
    tag = 'contextra-analytics:medium-saturation-19c6353-' + digest
    snap = R / 'snapshot'
    snap.mkdir()
    archive_process = subprocess.Popen(['git','archive',p.read('plan.json')['head']], stdout=subprocess.PIPE)
    with tarfile.open(fileobj=archive_process.stdout,mode='r|') as archive:
        archive.extractall(snap,filter='data')
    assert archive_process.wait() == 0
    for name in FILES:
        shutil.copyfile(REPO / name,snap / name)
    p.save('image-start.json',dict(tag=tag,hashes=hashes,at=time.time()))
    started = time.monotonic()
    with (R / 'image-build.log').open('x') as log:
        child = subprocess.Popen(['docker','build','--pull=false','--network=default','-f',
            'docker/Dockerfile','--target','app','-t',tag,'.'],cwd=snap,
            env=dict(os.environ,DOCKER_BUILDKIT='1'),stdout=log,stderr=log,start_new_session=True)
        while child.poll() is None:
            if time.monotonic()-started > 600 or shutil.disk_usage('/var/lib/docker').free < 15*1024**3:
                os.killpg(child.pid,signal.SIGTERM)
                child.wait(20)
                raise RuntimeError('BUILD_BOUND')
            time.sleep(2)
    assert child.returncode == 0
    image = json.loads(p.cmd(['docker','image','inspect',tag]))[0]['Id']
    p.save('image.json',dict(tag=tag,image=image,hashes=hashes,elapsed=time.monotonic()-started))
    p.tests('saturation-image',image)
    print('Image and targeted tests PASS',flush=True)


def configure(enabled):
    values = rt.states()
    for sid in rt.ORIGINAL:
        q = values[sid]
        if q['state'] == 'running':
            rt.api('/api/v1/streams/' + sid + '/metrics','POST',dict(
                run_generation=q['run_generation'],enabled=enabled,seconds=60))


def snap():
    values = rt.states()
    at = time.monotonic()
    channels = {}
    for row in previous.ROWS:
        sid = row['stream_id']; q = values[sid]; path = q['metadata_path']; perf = path['performance']
        channels[sid] = dict(state=q['state'],run_id=q['run_id'],generation=q['run_generation'],
            error=q.get('last_error'),processed=path.get('processed',0),queued=path.get('queued',0),
            stale=path.get('stale',0),drop=path.get('queue_drop',0),
            collector=path.get('collector_frames'),performance=perf,
            model=q['inference_profile']['active_models'],redis=rt.redis_sample(row))
    return dict(at=at,channels=channels,publisher=rt.api('/api/v1/metadata/status')['metrics'],resource=rt.resource())


def summary(label,samples):
    a,b = samples[0],samples[-1]; dt = b['at']-a['at']; rows=[]
    for sid in rt.ORIGINAL:
        x,y = a['channels'][sid],b['channels'][sid]
        assert x['run_id'] == y['run_id'] and x['generation'] == y['generation']
        redis_fps = None
        if not x['redis'].get('missing') and not y['redis'].get('missing'):
            redis_fps=(y['redis']['entries']-x['redis']['entries'])/(y['redis']['at']-x['redis']['at'])
        counters=x['performance']['counts']; end=y['performance']['counts']
        rows.append(dict(phase=label,stream_id=sid,seconds=dt,
            infer_fps=(end.get('infer',0)-counters.get('infer',0))/dt,
            native_fps=(end.get('native',0)-counters.get('native',0))/dt,
            result_fps=(y['processed']-x['processed'])/dt,redis_fps=redis_fps,
            queue_start=x['queued'],queue_end=y['queued'],queue_max=max(s['channels'][sid]['queued'] for s in samples),
            stale=y['stale']-x['stale'],drop=y['drop']-x['drop'],collector_start=x['collector'],collector_end=y['collector'],
            missed=y['performance']['observation_missed']-x['performance']['observation_missed'],
            redis_length=y['redis'].get('length'),redis_pttl=y['redis'].get('pttl')))
    p.save(label+'-summary.json',dict(rows=rows,details={s:b['channels'][s]['performance'] for s in rt.ORIGINAL},
        publisher_delta={k:b['publisher'].get(k,0)-a['publisher'].get(k,0) for k in ('sent_frames','replaced','expired','send_error')},
        resource_start=a['resource'],resource_end=b['resource']))
    print(label,[(r['stream_id'],round(r['result_fps'],2),r['queue_end'],r['stale'],r['collector_end']) for r in rows],flush=True)
    return rows


def unhealthy(samples):
    if len(samples)<3:
        return False
    for sid in rt.ORIGINAL:
        bad=[]
        for a,b in zip(samples[-3:-1],samples[-2:]):
            x,y=a['channels'][sid],b['channels'][sid]
            bad.append((y['processed']-x['processed'])/(b['at']-a['at'])<8
                       and y['stale']-x['stale']>=5 and y['queued']>0)
        if all(bad):
            return True
    return False


def observe():
    configure(True)
    time.sleep(5)
    samples=[snap()]; initial=[samples[0]]; phase='initial'; phase_start=samples[0]['at']; rows=[]
    last_first_frame=time.monotonic()-5
    probe_armed=False
    while True:
        time.sleep(5)
        current=snap(); samples.append(current)
        p.save('window.json',dict(samples=samples,phase=phase,complete=False))
        if any(q['state']!='running' or q['error'] for q in current['channels'].values()):
            raise RuntimeError('CHANNEL_FAILED')
        if unhealthy(samples):
            summary('deterioration',samples[-3:])
            raise RuntimeError('SUSTAINED_SLOWDOWN_NO_FURTHER_EXPERIMENT')
        if current['at']-last_first_frame>240:
            raise RuntimeError('SATURATION_BUDGET_EXPIRED')
        if phase=='initial':
            initial.append(current)
            if current['at']-phase_start>=20:
                rows+=summary('initial',initial)
                configure(False)
                phase='filling'
        elif phase=='filling':
            counts=[q['collector'] for q in current['channels'].values()]
            assert all(type(c) is int and 0<=c<=1000 for c in counts),'COLLECTOR_COUNT_MISSING'
            print('Collector counts',counts,flush=True)
            if min(counts)>=800 and not probe_armed:
                configure(True);probe_armed=True
            if min(counts)==1000:
                assert probe_armed
                phase='saturated';phase_start=current['at'];saturated=[current]
                p.save('saturation.json',dict(at=phase_start,counts=counts))
        elif phase=='saturated':
            saturated.append(current)
            assert all(q['collector']==1000 for q in current['channels'].values())
            if current['at']-phase_start>=30:
                rows+=summary('saturated',saturated)
                p.save('window.json',dict(samples=samples,phase=phase,complete=True))
                p.save('summary.json',rows)
                with (R/'summary.csv').open('w') as f:
                    writer=csv.DictWriter(f,fieldnames=list(rows[0]));writer.writeheader();writer.writerows(rows)
                return 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 for r in rows[-3:])


def trial():
    assert not (R/'trial-start.json').exists()
    plan=p.read('plan.json'); image=p.read('image.json')
    assert p.read('saturation-image.json')['success']
    assert p.sha(p.ACTIVE)==plan['active_hash'] and p.sha(p.ENV)==plan['env_hash']
    assert p.sha(p.prior.REG)==plan['registry_hash']
    assert all(p.sha(REPO/n)==v for n,v in image['hashes'].items())
    p.save('trial-start.json',dict(at=time.time(),image=image['image']))
    previous.stop_all('before-candidate')
    active=(R/'active.before').read_bytes();env=(R/'env.before').read_bytes()
    cfg=yaml.safe_load(active);cfg['services']['analytics']['image']=image['image']
    a=yaml.safe_dump(cfg).encode(); lines=env.decode().splitlines(keepends=True)
    idx=[i for i,l in enumerate(lines) if l.startswith('CONTEXTRA_IMAGE=')];assert len(idx)==1
    lines[idx[0]]='CONTEXTRA_IMAGE='+image['tag']+'\n';e=''.join(lines).encode()
    ah,eh=hashlib.sha256(a).hexdigest(),hashlib.sha256(e).hexdigest()
    p.save('mutation-intent.json',dict(active_after=ah,env_after=eh))
    success=False
    try:
        p.atomic(p.ACTIVE,a,plan['active_hash']);p.atomic(p.ENV,e,plan['env_hash'])
        previous.recreate(image['image'],'candidate')
        previous.start_all('candidate-start')
        success=observe()
    except Exception as error:
        p.save('failure.json',dict(type=type(error).__name__,reason=str(error)[:250],trace=traceback.format_exc()))
        print('TRIAL STOP',type(error).__name__,str(error)[:150],flush=True)
    finally:
        configure(False)
        if not success:
            previous.stop_all('restore-stop')
            p.atomic(p.ACTIVE,active,ah);p.atomic(p.ENV,env,eh)
            previous.recreate(plan['analytics']['Image'],'restored')
            previous.start_all('restored-start')
            time.sleep(5)
            previous.window('restored',5)
        current=p.base.inspect('contextra-analytics-1')
        assert current['Image']==(image['image'] if success else plan['analytics']['Image'])
        assert p.sha(p.ACTIVE)==(ah if success else plan['active_hash'])
        assert p.sha(p.ENV)==(eh if success else plan['env_hash'])
        assert p.sha(p.prior.REG)==plan['registry_hash']
        sender=p.prior.ops.prior.prev.sender()
        assert all(sender.get(k)==plan['sender'].get(k) for k in ('executionId','uri','video','mode','state'))
        assert p.base.inspect('contextra-redis-1')['Id']==plan['redis']
        values=rt.states()
        assert all(values[s]['state']=='running' and not values[s]['metadata_path']['performance']['detail'] for s in rt.ORIGINAL)
        p.save('final.json',dict(at=time.time(),success=success,image=current['Image'],states=values,
            preserved_settings=True,sender_preserved=True,redis_preserved=True,registry_unchanged=True,
            persistent_consistent=True,diagnostics_off=True,publisher=rt.api('/api/v1/metadata/status')))
        print('FINAL', 'candidate retained' if success else 'original restored',flush=True)


if __name__=='__main__':
    {'gate':gate,'test':lambda:p.tests('saturation-source'),'test2':lambda:p.tests('saturation-source2'),
     'build':build,'trial':trial}[sys.argv[1]]()