"""Local test-only APP patch. No remote access, channel command or enrollment."""
import copy
import hashlib
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time

REPO = Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
OLD = Path('/data/codex-artifacts/contextra/field-maintenance/2026/2026-10-21_ct-trace_r01/test')
ROOT = Path('/data/codex-artifacts/contextra/field-maintenance/2026/2026-10-21_trace-completion_r01')
definition = json.loads(Path('/etc/contextra-maintenance/server.json').read_text())
assert definition['server'] == 'contextra-local' and definition['project'] == 'contextra'
sys.path[:0] = [definition['paths']['tools'], str(REPO/'tools/icerti_offline')]
import compose_backend as cb
import field_profiles as fp
import server_context as server
from trace_install import command, sha, load
from trace_payload import function

STATE = cb.state_path()
TX = STATE/'transactions'/'trace-completion-20261021-r01'


def save(name, value): fp.atomic(ROOT/(name+'.json'), value)
def rows(): return {r['Config']['Labels']['com.docker.compose.service']: r for r in cb.containers()}
def py(cid, code): return command(['docker','exec','-i',cid,'python3','-B','-'],15,code.encode())
def disk():
    if shutil.disk_usage('/').free < 15*1024**3: raise ValueError('ROOT_RESERVE')
    server.guard_disks()


def idle(cid, require_idle=True):
    value = json.loads(py(cid, """import json,os,urllib.request
from pathlib import Path
t=Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip()
q=urllib.request.Request('http://127.0.0.1:5000/internal/secure-control/runtime-status',headers={'Authorization':'Bearer '+t})
with urllib.request.urlopen(q,timeout=5) as r: s=json.load(r)
import psycopg2
with psycopg2.connect(os.environ['DATABASE_URL'],connect_timeout=2,options='-c statement_timeout=1500 -c default_transaction_read_only=on') as c:
 with c.cursor() as q:
  q.execute("SELECT count(*) FROM contextra_event_artifacts WHERE artifact_state IN ('CAPTURING','RENDERING') OR delivery_state IN ('PENDING','RETRY','CLAIMED','SENDING')")
  n=q.fetchone()[0]
print(json.dumps({'channels':s['streamCounts'],'eventWork':n}))
"""))
    if require_idle and (any(value['channels'][k] for k in ('running','starting','error')) or value['eventWork']):
        raise ValueError('LIVE_WORK_REQUIRES_USER '+json.dumps(value))
    return value


def prepare():
    ROOT.mkdir(mode=0o700, parents=True, exist_ok=True)
    if (ROOT/'plan.json').exists(): raise ValueError('PRESERVE_PLAN')
    disk(); server.verify_tools()
    a,site,cp,approved,live,source = cb.inspect()
    if approved != live: raise ValueError('CONFIG_DRIFT')
    rs = rows(); expected = load(OLD/'verified.json')['services']
    if any(rs[k]['Image'] != v['image'] for k,v in expected.items()): raise ValueError('BASE_IMAGE_DRIFT')
    for name in ('pending.json','apply-pending.json','network-pending.json'):
        if (fp.STATE/name).exists(): raise ValueError('PENDING_OPERATION')
    activity = idle(rs['analytics']['Id'], require_idle=False) if rs['analytics']['State']['Running'] else 'stopped'
    manifest = load(OLD/'manifest.json')
    ctx = ROOT/'build'; payload = ctx/'payload'; payload.mkdir(parents=True)
    hashes = {}
    for name,h in manifest['after'].items():
        if not name.startswith(('app/','contextra_trace/')): continue
        source_file = OLD/'payload'/name
        if sha(source_file) != h: raise ValueError('OLD_PAYLOAD_DRIFT')
        dest = payload/name; dest.parent.mkdir(parents=True,exist_ok=True); shutil.copy2(source_file,dest)
        hashes[name] = h
    actual = json.loads(py(rs['analytics']['Id'], 'import json,hashlib\nfrom pathlib import Path\npaths='+repr(list(hashes))+"\nprint(json.dumps({p:hashlib.sha256((Path('/opt/contextra')/p).read_bytes()).hexdigest() for p in paths}))"))
    if actual != hashes: raise ValueError('LIVE_PAYLOAD_DRIFT')
    ctl = payload/'app/secure_control.py'; old = ctl.read_text()
    updated = old.replace(function(old,'completed'),function((REPO/'app/secure_control.py').read_text(),'completed'),1)
    compile(updated,str(ctl),'exec'); fp.atomic(ctl,updated.encode(),mode=0o644)
    shutil.copy2(REPO/'app/trace_commands.py',payload/'app/trace_commands.py')
    base = load(OLD/'built-analytics.json')['base']
    fp.atomic(ctx/'Dockerfile',('FROM '+base+'\nCOPY payload/ /opt/contextra/\n').encode(),mode=0o644)
    protected = [STATE/x for x in ('active.json','site.json','protected-bindings.json','input-baseline.json','trace/trace-policy.applied.json')]
    protected += [Path(definition['paths']['setup'])/'roi'/n for n in ('roi-policy.json','roi-policy.applied.json')]
    protected += [Path(definition['paths']['setup'])/'trace/trace-policy.json']
    save('plan',dict(active=a,sourceHash=cb.canonical(source),approved=approved,
        before={k:dict(id=r['Id'],image=r['Image'],running=r['State']['Running']) for k,r in rs.items()},
        protected={str(p):sha(p) for p in protected if p.exists()},
        oldPayload=hashes,newPayload={str(p.relative_to(payload)):sha(p) for p in payload.rglob('*.py')},
        base=base,activity=activity))
    shutil.copy2(__file__,ROOT/'deploy_completion.py')
    print(json.dumps({'phase':'PREPARED','activity':activity,'root':str(ROOT)}))


def guard():
    disk(); p=load(ROOT/'plan.json'); a,_,_,approved,live,source=cb.inspect(); rs=rows()
    if a!=p['active'] or approved!=p['approved'] or approved!=live or cb.canonical(source)!=p['sourceHash']: raise ValueError('CONFIG_CAS')
    if any(sha(path)!=h for path,h in p['protected'].items()): raise ValueError('PROTECTED_DRIFT')
    if any(rs[k]['Id']!=v['id'] or rs[k]['Image']!=v['image'] for k,v in p['before'].items()): raise ValueError('CONTAINER_CAS')
    if any(sha(ROOT/'build/payload'/name)!=h for name,h in p['newPayload'].items()): raise ValueError('PAYLOAD_DRIFT')
    return p,rs


def build():
    p,_=guard(); tag='contextra-trace:completion-test-20261021-r01'
    if (ROOT/'built.json').exists(): raise ValueError('PRESERVE_BUILT')
    with (ROOT/'build.log').open('ab') as log:
        r=subprocess.run(['docker','build','--pull=false','--network=none','--memory=2g','--cpu-period=100000','--cpu-quota=200000','-t',tag,str(ROOT/'build')],
                         stdout=log,stderr=subprocess.STDOUT,timeout=300,env=dict(os.environ,DOCKER_BUILDKIT='0'))
    if r.returncode: raise ValueError('BUILD_FAILED')
    image=json.loads(command(['docker','image','inspect',tag]))[0]
    old=json.loads(command(['docker','image','inspect',p['before']['analytics']['image']]))[0]
    if image['Config']!=old['Config']: raise ValueError('IMAGE_EXEC_CONFIG_CHANGED')
    if len(image['RootFS']['Layers'])!=len(old['RootFS']['Layers']): raise ValueError('IMAGE_LAYER_COUNT_CHANGED')
    name='ct-trace-completion-import-20261021'
    try:
        script="import app.secure_control, app.trace_commands; from concurrent.futures import Future; from types import SimpleNamespace; import logging; from app.multi_roi.projection import StartResult; from app.stream_lifecycle import StreamLifecycle; l=StreamLifecycle(); l.begin(); g=SimpleNamespace(lifecycle=l); f=Future(); f.set_result(StartResult(g,{})); rows=[]; app.trace_commands.completion(SimpleNamespace(operation_id='fixture',camera_id=3,action='start'),f,emit=lambda **r:rows.append(r),error_code=lambda e:'ERROR',logger=logging.getLogger('fixture')); assert rows[0]['runId']==l.run_id and rows[0]['phase']=='COMMAND_APPLIED'; print('ISOLATED_IMPORT_AND_COMPLETION_PASS')"
        out=command(['docker','run','--rm','--pull=never','--name',name,'--network=none','--read-only','--memory=512m','--cpus=1','--pids-limit=64','--tmpfs','/tmp:size=16m',
            '-e','NVIDIA_VISIBLE_DEVICES=void','-e','PYTHONDONTWRITEBYTECODE=1','-e','OPENBLAS_NUM_THREADS=1','--entrypoint','python3',image['Id'],'-c',script],30)
    finally:
        found=subprocess.run(['docker','inspect',name],capture_output=True,timeout=10)
        if found.returncode==0:
            row=json.loads(found.stdout)[0]
            if row['Image']!=image['Id']: raise ValueError('HELPER_IDENTITY')
            command(['docker','rm','-f',name],15)
    save('built',dict(image=image['Id'],tag=tag,layers=len(image['RootFS']['Layers']),check=out.decode().strip()))
    print((ROOT/'built.json').read_text())


def activate():
    p,rs=guard(); built=load(ROOT/'built.json'); app=rs['analytics']; gateway=rs['ai-gateway']
    if app['State']['Running']: idle(app['Id'])
    # Briefly hold ingress, then recheck; never send a channel Stop to win a race.
    if gateway['State']['Running']: command(['docker','stop','--signal','SIGTERM','--timeout','30',gateway['Id']],40)
    save('transition',dict(phase='GATEWAY_HELD',gatewayWasRunning=gateway['State']['Running']))
    try:
        time.sleep(3)
        if app['State']['Running']: idle(app['Id'])
    except Exception:
        if gateway['State']['Running']: command(['docker','start',gateway['Id']],30)
        raise
    if app['State']['Running']:
        command(cb.compose_args(p['active']['physical'])+['stop','--timeout','90','analytics'],110)
        stopped=rows()['analytics']
        if stopped['State']['Running'] or stopped['State']['ExitCode']!=0: raise ValueError('APP_SHUTDOWN_NOT_CLEAN')
    guard()
    new=copy.deepcopy(p['approved']);new['services']['analytics']['image']=built['image']
    TX.mkdir(mode=0o700)
    for name,v in [('compose.json',new),('approved.json',new),('compose.physical.json',server.physical(new)),('previous-active.json',p['active'])]: fp.atomic(TX/name,v)
    from apply_source import normalize
    rendered=json.loads(command(cb.compose_args(TX/'compose.physical.json')+['config','--format','json']))
    if normalize(server.logical(rendered))!=normalize(new): raise ValueError('COMPOSE_ROUNDTRIP')
    active=dict(p['active'],transaction=TX.name,compose=str(TX/'compose.json'),physical=str(TX/'compose.physical.json'),compose_semantic_sha256=cb.canonical(new))
    save('activation',dict(before=p['active'],after=active))
    if cb.active()!=p['active']: raise ValueError('ACTIVE_CAS')
    fp.atomic(STATE/'active.json',active)
    finish()


def finish():
    disk(); p=load(ROOT/'plan.json'); active=load(ROOT/'activation.json')['after']; rs=rows()
    if cb.active()!=active: raise ValueError('RESUME_ACTIVE_CAS')
    cb.inspect()
    if cb.canonical(cb.render())!=p['sourceHash']: raise ValueError('RESUME_SOURCE_DRIFT')
    if rs['analytics']['State']['Running'] or rs['ai-gateway']['State']['Running']: raise ValueError('RESUME_EXPECT_STOPPED')
    if rs['analytics']['Image']!=p['before']['analytics']['image']: raise ValueError('RESUME_APP_DRIFT')
    for k,v in p['before'].items():
        if rs[k]['Id']!=v['id']: raise ValueError('RESUME_CONTAINER_DRIFT')
    for path,h in p['protected'].items():
        if path!=str(STATE/'active.json') and sha(path)!=h: raise ValueError('RESUME_PROTECTED_DRIFT')
    command(cb.compose_args(active['physical'])+['up','--no-start','--no-deps','--no-build','--pull','never','analytics'],90)
    if p['before']['analytics']['running']: command(cb.compose_args(active['physical'])+['start','analytics'],90)
    save('deployed',dict(active=active,appWasRunning=p['before']['analytics']['running'],gatewayWasRunning=p['before']['ai-gateway']['running']))
    print('APP replaced; gateway held for readiness; no camera Start/Stop.')


def verify():
    p=load(ROOT/'plan.json'); d=load(ROOT/'deployed.json'); rs=rows(); built=load(ROOT/'built.json')
    if cb.active()!=d['active'] or rs['analytics']['Image']!=built['image']: raise ValueError('DEPLOY_DRIFT')
    if d['appWasRunning']:
        until=time.monotonic()+120
        while True:
            app=rows()['analytics']
            if app['State'].get('Health',{}).get('Status')=='healthy': break
            if not app['State']['Running'] or time.monotonic()>until: raise ValueError('APP_READINESS_FAILED')
            time.sleep(2)
        actual=json.loads(py(app['Id'],'import json,hashlib\nfrom pathlib import Path\npaths='+repr(list(p['newPayload']))+"\nprint(json.dumps({p:hashlib.sha256((Path('/opt/contextra')/p).read_bytes()).hexdigest() for p in paths}))"))
        if actual!=p['newPayload']: raise ValueError('INSTALLED_HASH_DRIFT')
    for path,h in p['protected'].items():
        if path!=str(STATE/'active.json') and sha(path)!=h: raise ValueError('PROTECTED_DRIFT')
    for role in ('ai-control-sidecar','ai-gateway','postgresql','redis'):
        if rs[role]['Id']!=p['before'][role]['id'] or rs[role]['Image']!=p['before'][role]['image']: raise ValueError('OTHER_SERVICE_CHANGED')
    if d['gatewayWasRunning']: command(['docker','start',rs['ai-gateway']['Id']],30)
    cb.inspect()
    result=dict(image=built['image'],payloadHashesVerified=True,otherContainersUnchanged=True,
                services={k:dict(running=r['State']['Running'],image=r['Image']) for k,r in rows().items()})
    save('verified',result); print(json.dumps(result))


if __name__=='__main__':
    {'prepare':prepare,'build':build,'activate':activate,'finish':finish,'verify':verify}[sys.argv[1]]()