"""Reviewed local-only deployment; stage transitions are explicit and resumable.""" import argparse import copy import hashlib import json import os from pathlib import Path import shutil import sys ROOT = Path(__file__).resolve().parent REPO = Path('/workspace/contextra-analytics-runtime-lifecycle-refactor') TOOLS = Path('/usr/local/lib/contextra-maintenance/portable-20261001T182138') STATE = Path('/workspace/contextra-runtime-config/ct/runtime/state') POLICY = Path('/workspace/contextra-runtime-config/ct/setup/roi/roi-policy.json') INSIDE = '/run/contextra-roi/roi-policy.json' HEAD = '3cca7d8ebf721ad6dd84e39ef3ce4a0c48527d7a' TX = STATE/'transactions'/'multi-roi-20261013-r01' DEFAULT = dict(mode='legacy', max_rois=8, max_rules=48, full_frame_roi_id=0) sys.path.insert(0, str(TOOLS)) import compose_backend as cb import field_profiles as fp import server_context as server from data_safety import lock from maintenance_process import run from deployment_kit import compose_snapshot, maintenance_bundle from apply_source import normalize def read(path): return json.loads(Path(path).read_text()) def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def command(args, timeout=30, input=None, env=None): return run(args, check=True, capture_output=True, text=True, timeout=timeout, input=input, cwd=REPO, env=env or compose_snapshot.command_environment()).stdout def out(value): print(json.dumps(value, ensure_ascii=False), flush=True) def receipt(phase, **fields): fp.atomic(ROOT/(phase+'.json'), dict(phase=phase, **fields)) out(dict(phase=phase, **fields)) def channels(): code = """import json,urllib.request,os,pathlib t=pathlib.Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip() r=urllib.request.Request('http://127.0.0.1:5000/internal/secure-control/runtime-status',headers={'Authorization':'Bearer '+t}) with urllib.request.urlopen(r,timeout=5) as s:d=json.load(s) print(json.dumps({'status':d['status'],'counts':d['streamCounts']})) """ return json.loads(command(['docker','exec','contextra-analytics-1','python3','-B','-c',code],10)) def assert_stopped(): value = channels() assert value['status'] == 'READY' assert value['counts']['stopped'] == value['counts']['total'], 'LIVE_CHANNEL_REQUIRES_NORMAL_STOP' return value def invariants(rows): result = {} for row in rows: role = row['Config']['Labels']['com.docker.compose.service'] result[role] = dict(id=row['Id'], image=row['Image'], started=row['State']['StartedAt'], mounts=cb.canonical(sorted(row['Mounts'],key=lambda m:m['Destination'])), restart=row['HostConfig']['RestartPolicy']) return result def preflight(): assert os.geteuid() == 0 and server.project() == 'contextra' and cb.state_path() == STATE assert shutil.disk_usage('/').free >= 17*1024**3 assert command(['git','rev-parse','HEAD']).strip() == HEAD server.verify_tools(); server.guard_disks() for name in ('pending.json','apply-pending.json','network-pending.json'): assert not (fp.STATE/name).exists(), 'PENDING_OPERATION' active = cb.active(); doc = read(active['compose']) assert cb.canonical(doc) == active['compose_semantic_sha256'] assert doc == read(Path(active['compose']).with_name('approved.json')) assert server.physical(doc) == read(active['physical']) source = cb.render() assert cb.canonical(source) == cb.canonical(read(STATE/'input-baseline.json')) protected = read(STATE/'protected-bindings.json') template = str(REPO/'docker/secure-gateway.conf.template') older = command(['git','show',HEAD+'^:docker/secure-gateway.conf.template']).encode() assert protected[template] == hashlib.sha256(older).hexdigest() other = {k:v for k,v in protected.items() if k != template} compose_snapshot.verify_bindings(other) assert sha(template) == hashlib.sha256(command(['git','show',HEAD+':docker/secure-gateway.conf.template']).encode()).hexdigest() return active, doc, source, protected, older def prepare(): assert not (ROOT/'plan.json').exists(), 'PLAN_EXISTS' a,doc,source,protected,oldtemplate = preflight() rows=cb.containers(); before=invariants(rows) assert set(before)==set(server.ROLES) and len(rows)==5 assert all(row['State']['Running'] for row in rows) assert all(before[k]['image']==v['image'] for k,v in doc['services'].items()) count=assert_stopped() changed=command(['git','diff','--name-only',HEAD+'^',HEAD,'--','app']).splitlines() payloads={} for name in changed: p=REPO/name; old=ROOT/'baseline'/name prior=run(['git','show',HEAD+'^:'+name],capture_output=True,timeout=10,cwd=REPO) if prior.returncode==0: assert old.read_bytes()==prior.stdout, 'APP_BASELINE_DRIFT:'+name else: assert not old.exists(), 'NEW_FILE_COLLISION' compile(p.read_bytes(),name,'exec') payloads[name]=sha(p) assert (ROOT/'baseline/sidecar-main.py').read_bytes()==command(['git','show',HEAD+'^:control_sidecar/main.py']).encode() for name in ('active.json','protected-bindings.json','input-baseline.json','site.json'): shutil.copy2(STATE/name,ROOT/'baseline'/name) fp.atomic(ROOT/'baseline/gateway-template.before',oldtemplate) # Preserve only non-secret structural evidence; existing private Compose stays in its original transaction. plan=dict(head=HEAD, active_before=a, active_hash=sha(STATE/'active.json'), protected_hash=sha(STATE/'protected-bindings.json'), input_hash=cb.canonical(source), compose_hash=sha(a['compose']), physical_hash=sha(a['physical']), before=before, channels=count, payloads=payloads, sidecar_sha=sha(REPO/'control_sidecar/main.py'), template_sha=sha(REPO/'docker/secure-gateway.conf.template'), unchanged_files={str(STATE/'site.json'):sha(STATE/'site.json'), str(STATE/'input-baseline.json'):sha(STATE/'input-baseline.json'), str(server.CONFIG):sha(server.CONFIG)}) fp.atomic(ROOT/'plan.json',plan) receipt('prepared', app_files=len(changed), channels=count, base_images={k:v['image'] for k,v in before.items()}) def build(): p=read(ROOT/'plan.json') assert not (ROOT/'built.json').exists(), 'BUILD_EXISTS' assert shutil.disk_usage('/').free >= 17*1024**3 images={} for role in ('analytics','ai-control-sidecar'): context=ROOT/('build-'+role); context.mkdir(mode=0o700,exist_ok=False) payload=context/'payload'; payload.mkdir() if role=='analytics': for name,expected in p['payloads'].items(): assert sha(REPO/name)==expected dest=payload/name; dest.parent.mkdir(parents=True,exist_ok=True); shutil.copy2(REPO/name,dest) copy_line='COPY payload/app/ /opt/contextra/app/' else: assert sha(REPO/'control_sidecar/main.py')==p['sidecar_sha'] shutil.copy2(REPO/'control_sidecar/main.py',payload/'main.py') copy_line='COPY payload/main.py /opt/bridge/control_sidecar/main.py' base=p['before'][role]['image'] fp.atomic(context/'Dockerfile',(f'FROM {base}\n{copy_line}\n').encode(),mode=0o600) tag='contextra-'+role+':multi-roi-3cca7d8-20261013-r01' env=compose_snapshot.command_environment() | {'DOCKER_BUILDKIT':'0'} result=run(['docker','build','--pull=false','--network','none','--memory','3g', '--cpu-period','100000','--cpu-quota','200000','-t',tag,str(context)], timeout=240,check=True,capture_output=True,text=True,env=env) fp.atomic(ROOT/('build-'+role+'.log'),(result.stdout+result.stderr).encode()) info=json.loads(command(['docker','image','inspect',tag]))[0] old=json.loads(command(['docker','image','inspect',base]))[0] assert info['RootFS']['Layers'][:len(old['RootFS']['Layers'])]==old['RootFS']['Layers'] images[role]=dict(id=info['Id'],tag=tag,base=base) out(dict(built=role,image=info['Id'])) receipt('built',images=images) def smoke(): built=read(ROOT/'built.json') for role,image in built['images'].items(): code=("import app.secure_control,app.multi_roi.runtime,app.multi_roi.artifacts; from app.multi_roi.policy import PolicyStore; assert PolicyStore().value['mode']=='legacy'; print('APP_IMPORT_LEGACY_OK')" if role=='analytics' else "import control_sidecar.main; print('SIDECAR_IMPORT_OK')") name='ct-roi-smoke-'+role try: result=command(['docker','run','--rm','--name',name,'--label','ct.roi-smoke=20261013-r01', '--network','none','--memory','1g','--cpus','1','--pids-limit','128','--read-only','--tmpfs','/tmp', '-e','PYTHONDONTWRITEBYTECODE=1','-e','OPENBLAS_NUM_THREADS=1', '--entrypoint','python3',image['id'],'-B','-c',code],45) out(dict(role=role,result=result.strip())) finally: ids=command(['docker','ps','-aq','--filter','name=^/'+name+'$','--filter','label=ct.roi-smoke=20261013-r01']).split() for ident in ids: command(['docker','rm','-f',ident],15) receipt('smoke',imports=True,network='none',gpu=False) def tools_patch(): p=read(ROOT/'plan.json'); server.verify_tools() assert not (ROOT/'tools-installed.json').exists() import portable_patch compatibility=portable_patch.compatibility() assert compatibility['images']=={k:v['image'] for k,v in p['before'].items()} result=maintenance_bundle.create(REPO/'tools/icerti_offline/maintenance',TOOLS,ROOT/'tools.zip', kind='TOOLS',version='multi-roi-20261013-r01',compatibility=compatibility, paths=['ct.py','field_roi.py'],changes=['MROI-01 policy CLI, no service action']) manifest,payload=maintenance_bundle.verify(result['bundle'],result['sha256']) journal=fp.STATE/'patches'/(result['sha256']+'.json') maintenance_bundle.apply(manifest,payload,TOOLS,journal,fp.atomic) portable_patch.update_inventory(TOOLS,manifest['files']); server.verify_tools() receipt('tools-installed',bundle=result,journal=str(journal),files=list(manifest['files'])) def stage(): p=read(ROOT/'plan.json'); built=read(ROOT/'built.json') assert read(ROOT/'smoke.json')['imports'] a,doc,source,protected,_=preflight() assert a==p['active_before'] and sha(STATE/'active.json')==p['active_hash'] assert not TX.exists() and not POLICY.parent.exists(), 'NEW_PATH_COLLISION' TX.mkdir(mode=0o700); POLICY.parent.mkdir(mode=0o700) fp.atomic(POLICY,DEFAULT,user=True) fp.atomic(POLICY.with_name('roi-policy.applied.json'),DEFAULT) candidate=copy.deepcopy(doc) for role,info in built['images'].items():candidate['services'][role]['image']=info['id'] analytics=candidate['services']['analytics'] analytics['environment']['CONTEXTRA_ROI_POLICY_FILE']=INSIDE analytics['volumes'].append({'type':'bind','source':str(POLICY.parent),'target':'/run/contextra-roi','bind':{'create_host_path':False}}) newtemplate=TX/'secure-gateway.conf.template' shutil.copy2(REPO/'docker/secure-gateway.conf.template',newtemplate) for volume in candidate['services']['ai-gateway']['volumes']: if volume['target']=='/etc/bridge/nginx.conf.template': oldtemplate=volume['source']; volume['source']=str(newtemplate) updated=copy.deepcopy(protected); updated.pop(oldtemplate); updated[str(newtemplate)]=sha(newtemplate) compose_snapshot.verify_bindings(updated) fp.atomic(TX/'compose.json',candidate); fp.atomic(TX/'approved.json',candidate) fp.atomic(TX/'compose.physical.json',server.physical(candidate)) actual=json.loads(command(cb.compose_args(TX/'compose.physical.json')+['config','--format','json'])) assert normalize(server.logical(actual))==normalize(candidate), 'COMPOSE_ROUNDTRIP' # Only reviewed differences are admitted; data services and all other config remain exact. expected=copy.deepcopy(candidate) for role in built['images']:expected['services'][role]['image']=doc['services'][role]['image'] expected['services']['analytics']['environment'].pop('CONTEXTRA_ROI_POLICY_FILE') expected['services']['analytics']['volumes'].pop() for volume in expected['services']['ai-gateway']['volumes']: if volume['target']=='/etc/bridge/nginx.conf.template':volume['source']=oldtemplate assert expected==doc after=dict(a,transaction=TX.name,compose=str(TX/'compose.json'),physical=str(TX/'compose.physical.json'),compose_semantic_sha256=cb.canonical(candidate)) fp.atomic(TX/'previous-active.json',a); fp.atomic(TX/'previous-bindings.json',protected) fp.atomic(TX/'next-active.json',after); fp.atomic(TX/'next-bindings.json',updated) receipt('staged',transaction=TX.name,policy=str(POLICY),mode='legacy',active_changed=False) def activate(): p=read(ROOT/'plan.json') assert (ROOT/'staged.json').exists() and (ROOT/'tools-installed.json').exists() with lock(fp.STATE/'data-maintenance.lock'): assert sha(STATE/'active.json')==p['active_hash'] assert sha(STATE/'protected-bindings.json')==p['protected_hash'] assert sha(p['active_before']['compose'])==p['compose_hash'] assert sha(p['active_before']['physical'])==p['physical_hash'] assert cb.canonical(cb.render())==p['input_hash'] server.guard_disks(); server.verify_tools() assert_stopped() receipt('transition',status='STOPPING_THREE_SERVICES',channels_start_allowed=False) # Remove incoming control first. SIGTERM is Nginx fast shutdown, not SIGKILL; # existing idle WSS sessions must not hold SIGQUIT until Compose kills it. command(['docker','stop','--signal','SIGTERM','--timeout','30','contextra-ai-gateway-1'],40) command(['docker','stop','--timeout','20','contextra-ai-control-sidecar-1'],30) result=run(['docker','exec','-i','contextra-analytics-1','python3','-B','-'], input=(TOOLS/'stop_channels.py').read_bytes(),check=True,capture_output=True,timeout=240) fp.atomic(ROOT/'stop-channels.log',result.stdout) command(cb.compose_args(p['active_before']['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-control-sidecar','ai-gateway'): assert not row['State']['Running'] and not row['State']['OOMKilled'] and row['State']['ExitCode']!=137, 'UNSAFE_STOP:'+role else:assert row['Id']==p['before'][role]['id'] and row['State']['Running'] receipt('services-stopped',codes={r['Name']:r['State']['ExitCode'] for r in rows}) # Journal both intended writes before the protected pointer transition. fp.atomic(ROOT/'activation-journal.json',dict(before=p['active_before'],after=read(TX/'next-active.json'),phase='POINTERS')) fp.atomic(STATE/'protected-bindings.json',read(TX/'next-bindings.json')) fp.atomic(STATE/'active.json',read(TX/'next-active.json')) cb.inspect(); cb.guard_images(read(TX/'compose.json')) receipt('activated',mode='legacy',services_started=False) def start_services(): assert (ROOT/'activated.json').exists() with lock(fp.STATE/'data-maintenance.lock'): a,_,_,base,live,_=cb.inspect() assert a['transaction']==TX.name and base==live server.guard_disks(); cb.guard_images(base) command(cb.compose_args(a['physical'])+['up','-d','--no-deps','--pull','never','--no-build', 'analytics','ai-control-sidecar','ai-gateway'],180) receipt('services-started',camera_start=False) def verify(): p=read(ROOT/'plan.json'); built=read(ROOT/'built.json') a,_,_,base,live,source=cb.inspect() assert a['transaction']==TX.name and base==live assert cb.canonical(source)==p['input_hash'] for name,digest in p['unchanged_files'].items():assert sha(name)==digest, 'BASELINE_CHANGED:'+name assert sha(p['active_before']['compose'])==p['compose_hash'] assert sha(p['active_before']['physical'])==p['physical_hash'] server.verify_tools(); server.guard_disks() rows=cb.containers(); after=invariants(rows) for row in rows: role=row['Config']['Labels']['com.docker.compose.service'] assert row['State']['Running'] and row['State'].get('Health',{}).get('Status','healthy')=='healthy' assert row['State']['Restarting'] is False and row['State']['OOMKilled'] is False assert after[role]['restart']==p['before'][role]['restart'] if role in ('redis','postgresql'): assert after[role]==p['before'][role], 'DATA_SERVICE_CHANGED' elif role in built['images']:assert row['Image']==built['images'][role]['id'] check = """import hashlib,json,os,pathlib,urllib.request,urllib.error from app.multi_roi.policy import PolicyStore p=pathlib.Path(os.environ['CONTEXTRA_ROI_POLICY_FILE']); store=PolicyStore(p) token=pathlib.Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip() url='http://127.0.0.1:5000/internal/secure-control/roi-policy' req=urllib.request.Request(url,headers={'Authorization':'Bearer '+token}) with urllib.request.urlopen(req,timeout=5) as response: status=json.load(response) assert status['configured']==status['effective']==store.value=={'mode':'legacy','max_rois':8,'max_rules':48,'full_frame_roi_id':0} assert status['pending'] is False and status['error'] is None codes={} for method in ('GET','PATCH'): try: urllib.request.urlopen(urllib.request.Request(url,method=method,data=b'{}' if method=='PATCH' else None),timeout=5) raise AssertionError('UNAUTHENTICATED_POLICY_ALLOWED') except urllib.error.HTTPError as e: codes[method]=e.code; assert e.code==401 manifest=json.load(__import__('sys').stdin) for name,expected in manifest.items(): actual=hashlib.sha256((pathlib.Path('/opt/contextra')/name).read_bytes()).hexdigest() assert actual==expected, name print(json.dumps({'configured':status['configured'],'effective':status['effective'],'fresh_load_equal':True,'auth_denials':codes,'verified_app_files':len(manifest)})) """ checks=json.loads(command(['docker','exec','-i','contextra-analytics-1','python3','-B','-c',check],15,input=json.dumps(p['payloads']))) sidecar=command(['docker','exec','contextra-ai-control-sidecar-1','python3','-B','-c', "import hashlib,pathlib; print(hashlib.sha256(pathlib.Path('/opt/bridge/control_sidecar/main.py').read_bytes()).hexdigest())"]).strip() assert sidecar==p['sidecar_sha'] command(['docker','exec','contextra-ai-gateway-1','nginx','-t','-c','/tmp/nginx.conf'],10) assert sha(TX/'secure-gateway.conf.template')==p['template_sha'] assert read(POLICY)==read(POLICY.with_name('roi-policy.applied.json'))==DEFAULT for path in (POLICY,POLICY.with_name('roi-policy.applied.json')): assert path.stat().st_uid==0 and path.stat().st_mode & 0o777 == 0o600 stopped=assert_stopped() assert stopped['counts']==p['channels']['counts'] assert shutil.disk_usage('/').free>=15*1024**3 receipt('verified',checks=checks,channels=stopped,data_services_unchanged=True, restart_policies_preserved=True,old_transactions_preserved=True, images={k:v['image'] for k,v in after.items()}, root_free_gib=round(shutil.disk_usage('/').free/1024**3,2)) def rollback_plan(): p=read(ROOT/'plan.json') target=STATE/'transactions'/'multi-roi-20261013-r01-rollback' assert not target.exists() target.mkdir(mode=0o700) doc=read(p['active_before']['compose']) template=target/'secure-gateway.conf.template' shutil.copy2(ROOT/'baseline/gateway-template.before',template) for v in doc['services']['ai-gateway']['volumes']: if v['target']=='/etc/bridge/nginx.conf.template': old=v['source']; v['source']=str(template) protected=read(ROOT/'baseline/protected-bindings.json') assert sha(template)==protected.pop(old) protected[str(template)]=sha(template) compose_snapshot.verify_bindings(protected) fp.atomic(target/'compose.json',doc); fp.atomic(target/'approved.json',doc) fp.atomic(target/'compose.physical.json',server.physical(doc)) actual=json.loads(command(cb.compose_args(target/'compose.physical.json')+['config','--format','json'])) assert normalize(server.logical(actual))==normalize(doc) active=dict(p['active_before'],transaction=target.name,compose=str(target/'compose.json'), physical=str(target/'compose.physical.json'),compose_semantic_sha256=cb.canonical(doc)) fp.atomic(target/'rollback-active.json',active); fp.atomic(target/'rollback-bindings.json',protected) receipt('rollback-prepared',transaction=str(target),executed=False, note='Explicit follow-up approval and current-state CAS required; no ct config restore.') def main(): parser=argparse.ArgumentParser(); parser.add_argument('phase',choices=('prepare','build','smoke','tools','stage','activate','start','verify','rollback-plan')) phase=parser.parse_args().phase {'prepare':prepare,'build':build,'smoke':smoke,'tools':tools_patch,'stage':stage,'activate':activate,'start':start_services, 'verify':verify,'rollback-plan':rollback_plan}[phase]() if __name__=='__main__':main()