"""Test-only EVTIME-01 descendant image/transaction; never issues channel Start."""
import argparse
import ast
import copy
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time

ROOT=Path(__file__).resolve().parent
REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
prior=Path('/data/codex-artifacts/contextra/field-maintenance/2026/2026-10-13_test-multi-roi-install_r01/install.py')
spec=importlib.util.spec_from_file_location('prior_install',prior)
old=importlib.util.module_from_spec(spec);spec.loader.exec_module(old)
cb,fp,server=old.cb,old.fp,old.server
command,read,sha=old.command,old.read,old.sha
FILES=['app/contracts/frames.py','app/frame_time.py','app/event_capture.py',
       'app/event_artifact_contract.py','app/event_time_fallback.py']
TAG='contextra-analytics:test-event-time-d28a00d-20261013-r01'
TX=old.STATE/'transactions'/'event-time-fallback-20261013-r01'

def save(name,value):
    fp.atomic(ROOT/(name+'.json'),value)

def app(code,timeout=15):
    return json.loads(command(['docker','exec','-i','contextra-analytics-1','python3','-B','-'],
                             timeout,input=code))

def hashes():
    return app("import json,hashlib;from pathlib import Path;print(json.dumps({str(p.relative_to('/opt/contextra')):hashlib.sha256(p.read_bytes()).hexdigest() for p in Path('/opt/contextra/app').rglob('*.py')}))")

def runtime():
    return app("""import json,os,hashlib,urllib.request
from pathlib import Path
t=Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip()
def get(p):
 with urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:5000'+p,headers={'Authorization':'Bearer '+t}),timeout=5) as r:return json.load(r)
out={'channels':get('/internal/secure-control/runtime-status'),'policy':get('/internal/secure-control/roi-policy')}
out['config_hashes']={}
for k,d in [('STREAMS_CONFIG_PATH','/etc/contextra/streams.yaml'),('STREAMS_RUNTIME_PATH','/var/lib/contextra/runtime/streams_runtime.yaml')]:
 p=Path(os.environ.get(k,d));out['config_hashes'][k]=hashlib.sha256(p.read_bytes()).hexdigest() if p.exists() else None
print(json.dumps(out))
""")

def guards(plan):
    server.verify_tools();server.guard_disks()
    a,_,_,approved,live,source=cb.inspect()
    assert a==plan['active'] and approved==live
    assert cb.canonical(source)==plan['source_stamp']
    assert sha(old.STATE/'active.json')==plan['active_hash']
    for path,h in plan['protected_files'].items():assert sha(path)==h,path
    for name,h in plan['payloads'].items():assert sha(REPO/name)==h,name
    assert shutil.disk_usage('/').free>15*1024**3
    return a,approved

def prepare():
    os.chmod(ROOT,0o700)
    assert not (ROOT/'plan.json').exists()
    server.verify_tools();server.guard_disks()
    assert server.project()=='contextra' and cb.state_path()==old.STATE
    for name in ('pending.json','apply-pending.json','network-pending.json'):
        assert not (fp.STATE/name).exists(),name
    a,_,_,approved,live,source=cb.inspect();assert approved==live
    rows=cb.containers();before=old.invariants(rows)
    assert len(rows)==5 and all(r['State']['Running'] for r in rows)
    assert shutil.disk_usage('/').free>17*1024**3
    observed=hashes()
    wanted={str(p.relative_to(REPO)):sha(p) for p in (REPO/'app').rglob('*.py')}
    differences=sorted(k for k in set(observed)|set(wanted) if observed.get(k)!=wanted.get(k))
    assert differences==sorted(FILES),differences
    for name in FILES:
        parent=subprocess.run(['git','show','d28a00d^:'+name],cwd=REPO,capture_output=True,timeout=10)
        if parent.returncode==0:
            assert observed[name]==hashlib.sha256(parent.stdout).hexdigest(),name
            dest=ROOT/'baseline'/name;dest.parent.mkdir(parents=True,exist_ok=True);dest.write_bytes(parent.stdout)
        else:assert name not in observed
    paths=[old.STATE/n for n in ('site.json','input-baseline.json','protected-bindings.json')]
    paths += [server.CONFIG,old.POLICY,old.POLICY.with_name('roi-policy.applied.json')]
    r=runtime();assert r['policy']['configured']==r['policy']['effective'] and not r['policy']['pending']
    plan=dict(active=a,active_hash=sha(old.STATE/'active.json'),source_stamp=cb.canonical(source),
              before=before,protected_files={str(p):sha(p) for p in paths},runtime=r,
              payloads={p:wanted[p] for p in FILES},all_app_hashes=wanted,
              head=command(['git','rev-parse','HEAD']).strip())
    save('plan',plan)
    print(json.dumps({'phase':'prepared','files':differences,'mode':r['policy']['effective'],
                      'channels':r['channels']['streamCounts'],'base_image':before['analytics']['image']}))

def build():
    p=read(ROOT/'plan.json');guards(p)
    assert not (ROOT/'built.json').exists()
    context=ROOT/'build-context';context.mkdir(mode=0o700)
    for name in FILES:
        dest=context/'payload'/name;dest.parent.mkdir(parents=True,exist_ok=True)
        shutil.copy2(REPO/name,dest)
    selected=['test_event_time_fallback','test_frame_time','test_multi_roi_runtime']
    needed=set(selected);done=set();testdir=context/'tests';testdir.mkdir()
    while needed-done:
        name=sorted(needed-done)[0];path=REPO/'tests'/(name+'.py');body=path.read_text()
        shutil.copy2(path,testdir/path.name);done.add(name)
        for n in ast.walk(ast.parse(body)):
            if isinstance(n,ast.ImportFrom) and n.module:
                if n.module.startswith('tests.'):needed.add(n.module.split('.')[1])
                elif n.module=='tests':needed.update(x.name for x in n.names)
                elif (REPO/'tests'/(n.module+'.py')).is_file():needed.add(n.module)
    base=p['before']['analytics']['image']
    dockerfile=[f'FROM {base} AS verify','WORKDIR /opt/contextra','COPY payload/app/ /opt/contextra/app/',
        'COPY tests/ /opt/contextra/tests/',
        'RUN PYTHONDONTWRITEBYTECODE=1 OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 timeout 90 python3 -m pytest -q -p no:cacheprovider '+
        ' '.join('tests/'+x+'.py' for x in selected),f'FROM {base}']
    for name in FILES:dockerfile.append(f'COPY --from=verify /opt/contextra/{name} /opt/contextra/{name}')
    fp.atomic(context/'Dockerfile',('\n'.join(dockerfile)+'\n').encode())
    env=old.compose_snapshot.command_environment()|{'DOCKER_BUILDKIT':'0'}
    with (ROOT/'build.log').open('xb') as log:
        run=subprocess.run(['docker','build','--pull=false','--network','none','--memory','3g',
            '--cpu-period','100000','--cpu-quota','200000','-t',TAG,str(context)],
            stdout=log,stderr=subprocess.STDOUT,timeout=300,env=env)
    print((ROOT/'build.log').read_text()[-4500:]);assert run.returncode==0
    image=json.loads(command(['docker','image','inspect',TAG]))[0]
    baseinfo=json.loads(command(['docker','image','inspect',base]))[0]
    assert image['RootFS']['Layers'][:len(baseinfo['RootFS']['Layers'])]==baseinfo['RootFS']['Layers']
    assert image['Config']==baseinfo['Config'],'IMAGE_EXECUTION_CONFIG_CHANGED'
    save('built',dict(image=image['Id'],tag=TAG,base=base))

def deploy():
    p=read(ROOT/'plan.json');built=read(ROOT/'built.json')
    with old.lock(fp.STATE/'data-maintenance.lock'):
        a,doc=guards(p)
        assert runtime()['config_hashes']==p['runtime']['config_hashes'],'CAMERA_CONFIG_DRIFT'
        save('transition',{'phase':'STOPPING','agent_channel_start':False})
        command(['docker','stop','--signal','SIGTERM','--timeout','30','contextra-ai-gateway-1'],40)
        result=old.run(['docker','exec','-i','contextra-analytics-1','python3','-B','-'],
            input=(old.TOOLS/'stop_channels.py').read_bytes(),capture_output=True,check=True,timeout=240)
        fp.atomic(ROOT/'stop-channels.log',result.stdout)
        old.assert_stopped()
        save('ledger-before',ledger())
        command(cb.compose_args(a['physical'])+['stop','--timeout','90','analytics'],110)
        rows=cb.containers()
        for row in rows:
            role=row['Config']['Labels']['com.docker.compose.service']
            if role in ('analytics','ai-gateway'):
                assert not row['State']['Running'] and not row['State']['OOMKilled'] and row['State']['ExitCode']!=137
            else:assert row['Id']==p['before'][role]['id'] and row['State']['Running']
        guards(p)
        new=copy.deepcopy(doc);new['services']['analytics']['image']=built['image']
        inverse=copy.deepcopy(new);inverse['services']['analytics']['image']=doc['services']['analytics']['image'];assert inverse==doc
        TX.mkdir(mode=0o700);fp.atomic(TX/'compose.json',new);fp.atomic(TX/'approved.json',new)
        fp.atomic(TX/'compose.physical.json',server.physical(new));fp.atomic(TX/'previous-active.json',a)
        reread=json.loads(command(cb.compose_args(TX/'compose.physical.json')+['config','--format','json']))
        assert old.normalize(server.logical(reread))==old.normalize(new)
        updated=dict(a,transaction=TX.name,compose=str(TX/'compose.json'),physical=str(TX/'compose.physical.json'),
                     compose_semantic_sha256=cb.canonical(new))
        fp.atomic(TX/'hotfix.json',dict(implementation='d28a00d',image=built,payloads=p['payloads'],parent=a))
        save('activation',dict(before=a,after=updated,phase='POINTER_CHANGE'))
        assert cb.active()==a;fp.atomic(old.STATE/'active.json',updated)
        cb.inspect();cb.guard_images(new)
        command(cb.compose_args(updated['physical'])+['up','-d','--no-deps','--pull','never','--no-build','analytics'],150)
        save('deployed',dict(active=updated,image=built['image'],gateway='STOPPED_PENDING_READY'))
        print('Analytics replaced; Gateway held until readiness. No channel Start.')

def ledger(cutoff=None):
    return app('CUTOFF='+repr(cutoff)+'\n'+r'''
import os,json,time,hashlib,psycopg2
c=psycopg2.connect(os.environ['DATABASE_URL'],connect_timeout=4,options='-c statement_timeout=8000');c.set_session(readonly=True)
cutoff=time.time() if CUTOFF is None else CUTOFF
with c.cursor() as q:
 q.execute("SELECT row_to_json(t)::text FROM contextra_event_artifacts t WHERE delivery_state='BLOCKED_TIME' AND updated_at<=%s ORDER BY event_id",(cutoff,))
 h=hashlib.sha256();n=0
 for row in q:h.update(row[0].encode()+b'\n');n+=1
c.rollback();c.close()
print(json.dumps(dict(cutoff=cutoff,rows=n,sha256=h.hexdigest())))
''',20)

def verify():
    p=read(ROOT/'plan.json');built=read(ROOT/'built.json')
    assert cb.active()['transaction']==TX.name
    deadline=time.monotonic()+120
    while True:
        try:
            ready=app("import json,urllib.request;print(urllib.request.urlopen('http://127.0.0.1:5000/ready',timeout=3).read().decode())",6)
            if ready.get('ready'):break
        except subprocess.CalledProcessError:
            pass
        if time.monotonic()>=deadline:raise TimeoutError('API_READINESS_120S')
        time.sleep(2)
    r=runtime();assert r['policy']==p['runtime']['policy'],'POLICY_CHANGED'
    assert r['config_hashes']==p['runtime']['config_hashes'],'CAMERA_CONFIG_CHANGED'
    for path,h in p['protected_files'].items():assert sha(path)==h,path
    assert cb.canonical(cb.inspect()[-1])==p['source_stamp']
    assert hashes()==p['all_app_hashes'],'APP_SOURCE_MISMATCH'
    before=read(ROOT/'ledger-before.json');after=ledger(before['cutoff']);assert before==after
    checks=app("""import json,urllib.request,urllib.error,os
from pathlib import Path
out={}
for name in ('RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke','Awiros_person_and_head_detection'):
 with urllib.request.urlopen('http://127.0.0.1:8000/v2/models/'+name+'/ready',timeout=5) as r:out[name]=r.status
with urllib.request.urlopen('http://127.0.0.1:5000/ready',timeout=5) as r:out['api']=json.load(r)
try:urllib.request.urlopen('http://127.0.0.1:5000/internal/secure-control/runtime-status',timeout=5)
except urllib.error.HTTPError as e:out['unauthorized_status']=e.code
print(json.dumps(out))
""")
    assert checks['api']['ready'] and checks['unauthorized_status']==401
    rows=cb.containers()
    for row in rows:
        role=row['Config']['Labels']['com.docker.compose.service'];b=p['before'][role]
        assert row['HostConfig']['RestartPolicy']==b['restart']
        assert cb.canonical(sorted(row['Mounts'],key=lambda m:m['Destination']))==b['mounts']
        assert row['Image']==(built['image'] if role=='analytics' else b['image'])
        if role not in ('analytics','ai-gateway'):
            assert row['Id']==b['id'] and row['State']['StartedAt']==b['started']
    old.assert_stopped()
    command(['docker','start','contextra-ai-gateway-1'],40)
    time.sleep(3)
    command(['docker','exec','contextra-ai-gateway-1','nginx','-t','-c','/tmp/nginx.conf'],10)
    sidecode="""import json,os,urllib.request
from pathlib import Path
t=Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip();out={}
for path in ('/ready','/internal/bridge-status'):
 req=urllib.request.Request('http://127.0.0.1:8080'+path,headers={'Authorization':'Bearer '+t})
 with urllib.request.urlopen(req,timeout=5) as r:out[path]=json.load(r)
print(json.dumps(out))
"""
    side=json.loads(command(['docker','exec','contextra-ai-control-sidecar-1','python3','-c',sidecode],15))
    assert side['/ready']['ready'] and side['/internal/bridge-status']['auth'] is None
    final=runtime()
    result=dict(image=built,transaction=TX.name,checks=checks,sidecar=side,policy=final['policy'],
                channels=final['channels']['streamCounts'],blocked_rows_unchanged=before,
                gateway='RUNNING',agent_channel_start=False,external_start_allowed=True,
                root_free_bytes=shutil.disk_usage('/').free)
    save('verified',result);fp.atomic(TX/'verification.json',result)
    print(json.dumps(result,indent=2))

if __name__=='__main__':
    parser=argparse.ArgumentParser();parser.add_argument('phase',choices=('prepare','build','deploy','verify'))
    globals()[parser.parse_args().phase]()