"""Bounded public-config rollout; only approved Analytics/Gateway transitions.""" import json from pathlib import Path import subprocess import sys import time from prior_install import FallR2Installer from trace_install import command, load, sha class PublicConfigInstaller(FallR2Installer): def cpu_job(self,name,image,arguments,timeout): name='ct-fall-config-1102-r01-'+name.replace('ct-fall-r2-','').replace('-20261101-r01','') if subprocess.run(['docker','inspect',name],capture_output=True,timeout=10).returncode==0: raise ValueError('TRIAL_NAME_OCCUPIED') self.guard_payload() try: return command(['docker','run','--rm','--pull=never','--name',name,'--user','0:0','--runtime','runc', '--label','contextra.config-trial='+self.root.name,'--network=none','--read-only','--cpus','2', '--memory','8g','--pids-limit','256','--cap-drop=ALL','--security-opt=no-new-privileges', '--tmpfs','/tmp:rw,size=256m','-e','NVIDIA_VISIBLE_DEVICES=void','-e','PYTHONDONTWRITEBYTECODE=1', '-e','OPENBLAS_NUM_THREADS=1','-v',str(self.root)+':/work','-v',str(self.root)+':/out', '-v','/storage/contextra/experiments/action-model-eval/2026:/evidence:ro', '--entrypoint','python3',image,'-B',*arguments],timeout) 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 or row['Config']['Labels'].get('contextra.config-trial')!=self.root.name: raise ValueError('TRIAL_IDENTITY_CHANGED') command(['docker','rm','-f',name],20) def dev(self): _,rows=self.guard() out=self.cpu_job('dev',rows['analytics']['Image'],['/work/test_runner.py','dev'],180) self.fp.atomic(self.root/'dev.log',out);print(out.decode()) def test(self): self.guard() out=self.cpu_job('site',load(self.root/'built.json')['image'],['/work/test_runner.py','site'],180) self.fp.atomic(self.root/'site.log',out) self.save('isolated-tests',load(self.root/'site-tests.json')) print(out.decode()) def replay(self): self.guard() out=self.cpu_job('replay',load(self.root/'built.json')['image'],['/work/replay.py'],600) self.fp.atomic(self.root/'replay.log',out) value=load(self.root/'replay_result.json') print(json.dumps({k:v for k,v in value.items() if k!='items'})) def activate(self): for name in ('dev-tests','site-tests'): value=load(self.root/(name+'.json')) if value['exit_code'] or value['failed'] or value['errors']:raise ValueError('CONFIG_TEST_GATE_FAILED') value=load(self.root/'replay_result.json') if value['status']!='PASS' or value['inputs']!=36 or value['displayed']!=20145: raise ValueError('R2_REPLAY_GATE_FAILED') _,rows=self.guard() current=json.loads(self.python(rows['analytics']['Id'], 'import json,hashlib\nfrom pathlib import Path\nnames='+repr(list(self.m['guards']))+ '\nprint(json.dumps({n:hashlib.sha256((Path("/opt/contextra")/n).read_bytes()).hexdigest() for n in names}))')) if current!=self.m['guards']:raise ValueError('UNRELATED_SOURCE_DRIFT') super().activate() def verify(self): plan=load(self.root/'plan.json');deployed=load(self.root/'deployed.json') if self.cb.active()!=deployed['active']:raise ValueError('ACTIVE_DRIFT') self.cb.inspect();rows=self.rows();row=rows['analytics'];deadline=time.monotonic()+120 while True: actual=json.loads(command(['docker','inspect',row['Id']]))[0] if actual['State'].get('Health',{}).get('Status')=='healthy':break if not actual['State']['Running'] or time.monotonic()>deadline:raise ValueError('READINESS_FAILED') time.sleep(2) if row['Image']!=load(self.root/'built.json')['image']:raise ValueError('IMAGE_MISMATCH') for role in ('ai-control-sidecar','redis','postgresql','ai-gateway'): if rows[role]['Id']!=plan['before'][role]['id'] or rows[role]['Image']!=plan['before'][role]['image']: raise ValueError('UNRELATED_CONTAINER_CHANGED') for path,expected in plan['protected'].items(): if path==str(self.state/'protected-bindings.json'):expected=load(self.root/'settings-commit.json')['bindings_after'] if sha(path)!=expected:raise ValueError('PROTECTED_CONFIG_CHANGED') for entry in load(self.root/'settings-migration.json'): if sha(entry['path'])!=entry['after']:raise ValueError('SETTINGS_CAS_DRIFT') expected=dict(self.m['guards'],**self.m['after']) current=json.loads(self.python(row['Id'], 'import json,hashlib\nfrom pathlib import Path\nnames='+repr(list(expected))+ '\nprint(json.dumps({n:hashlib.sha256((Path("/opt/contextra")/n).read_bytes()).hexdigest() for n in names}))')) if current!=expected:raise ValueError('SOURCE_HASH_MISMATCH') code="""import json,os,urllib.request,yaml from pathlib import Path from app.fall_runtime import FallSettings from app.fight_runtime import FightSettings from app.fall_r2.config import R2Config stored=yaml.safe_load(Path(os.environ['STREAMS_RUNTIME_PATH']).read_text())['added'];values=[] for spec in stored: with urllib.request.urlopen('http://127.0.0.1:5000/api/v1/streams/'+spec['stream_id']+'/settings',timeout=4) as r:d=json.load(r) expected=FallSettings.parse(spec.get('fall') or {}).settings_payload() for label in ('configured','effective'):assert d[label]['fall']==expected assert d['effective']['fight']==FightSettings.parse(spec.get('fight') or {}).settings_payload() values.append(dict(stream=spec['stream_id'],fall_engine=expected.get('engine','legacy'),r2=expected.get('r2'), enabled=expected['enabled'],person=expected['person_score'],joint=expected['joint_score'],cooldown=spec.get('event_cooldowns',{}).get('FALL'))) template=json.loads(Path(os.environ['CONTEXTRA_CONTROL_TEMPLATE_FILE']).read_text()) assert template['fall']['engine']=='r2' assert FallSettings.parse(template['fall']).r2.payload()==R2Config().payload() q=urllib.request.Request('http://127.0.0.1:8000/v2/repository/index',data=b'{}',headers={'Content-Type':'application/json'}) with urllib.request.urlopen(q,timeout=5) as r:models=json.load(r) assert {'RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke','Awiros_person_and_head_detection'}.issubset({x['name'] for x in models if x['state']=='READY'}) print(json.dumps(dict(settings=values,template=template['fall']['r2'],models_ready=True))) """ config=json.loads(self.python(row['Id'],code,120)) self.save('verified-before-gateway',dict(source_verified=True,**config,runtime=self.runtime(row['Id']))) if deployed['gatewayHeld']:command(['docker','start',rows['ai-gateway']['Id']],30) self.save('gateway-restored',dict(restored=deployed['gatewayHeld'])) time.sleep(5) final=self.runtime(row['Id']) self.save('verified',dict(status='AI_VERIFIED_BROWSER_NOT_TESTED',source_verified=True, channel_count=len(config['settings']),models_ready=True,gateway_restored=deployed['gatewayHeld'], runtime_counts=final['streamCounts'],camera_start_calls=0,**config)) logs=command(['docker','logs','--since','5m','--tail','300',row['Id']],30) self.fp.atomic(self.root/'restricted-post-deploy.log',logs) print(json.dumps({k:v for k,v in load(self.root/'verified.json').items() if k not in ('settings','template')})) if __name__=='__main__': root,expected,action=sys.argv[1:] if action not in ('prepare','dev','build','test','replay','activate','verify'):raise ValueError('ACTION_NOT_ALLOWED') getattr(PublicConfigInstaller(root,expected),action)()