"""R5-SITE-DEFAULT-01: Analytics-only rollout; no camera Start or automatic rollback.""" import copy import json import os from pathlib import Path import shutil import stat import subprocess import sys import time from artifact_install import ArtifactInstaller from trace_install import command, load, sha def host_path(row, target): matches = [m for m in row['Mounts'] if m['Type']=='bind' and (target==m['Destination'] or target.startswith(m['Destination']+'/'))] if not matches: raise ValueError('R5_CONFIG_BIND_REQUIRED') mount = max(matches, key=lambda m:len(m['Destination'])) return Path(mount['Source']) / Path(target).relative_to(mount['Destination']) def updated_bindings(records, changes): result = copy.deepcopy(records) for entry in changes: if not entry['changed']: continue path = Path(entry['path']) for parent, record in result.items(): if parent == str(path): if record != entry['before']: raise ValueError('R5_BINDING_CAS_FAILED') result[parent] = entry['after'] elif isinstance(record,dict) and path.is_relative_to(parent): key = str(path.relative_to(parent)) if record['directory'].get(key) != entry['before']: raise ValueError('R5_BINDING_MEMBER_CAS_FAILED') record['directory'][key] = entry['after'] return result class FightR5Installer(ArtifactInstaller): def __init__(self, root, expected): super().__init__(root, expected) self.tx = self.state/'transactions'/('fight-r5-'+self.root.name) def cpu_job(self, name, image, arguments, timeout): try: return command(['docker','run','--rm','--pull=never','--name',name,'--user','0:0','--runtime','runc', '--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', '--entrypoint','python3',image,'-B',*arguments],timeout) finally: result=subprocess.run(['docker','inspect',name],capture_output=True,timeout=10) if result.returncode==0: row=json.loads(result.stdout)[0] if row['Image']!=image:raise ValueError('R5_TRIAL_IDENTITY_CHANGED') command(['docker','rm','-f',name],20) def prepare(self): super().prepare() row=self.rows()['analytics'] env=dict(v.split('=',1) for v in row['Config']['Env'] if '=' in v) targets=[] for key,kind,fmt in (('CONTEXTRA_CONTROL_TEMPLATE_FILE','template','json'), ('STREAMS_CONFIG_PATH','base','yaml'),('STREAMS_RUNTIME_PATH','runtime','yaml')): path=self.s.safe(host_path(row,env[key])) if not path.is_file():raise ValueError('R5_CONFIG_FILE_REQUIRED') targets.append(dict(kind=kind,format=fmt,path=str(path),container_path=env[key],name=kind+'.'+fmt)) self.save('config-targets',targets) self.save('indexes',dict(schemaUnchanged=True,migrationRequired=False)) def test(self): self.guard() built=load(self.root/'built.json') name='ct-r5-config-test-20261101-r01' code="""import json,sys sys.path[:0]=['/work','/work/tests','/opt/contextra'] import pytest class Report: def __init__(self):self.passed=[];self.failed=[];self.errors=[];self.collected=0 def pytest_collection_finish(self,session):self.collected=len(session.items) def pytest_runtest_logreport(self,report): if report.failed:(self.failed if report.when=='call' else self.errors).append(report.nodeid) elif report.when=='call' and report.passed:self.passed.append(report.nodeid) r=Report();code=pytest.main(['-q','-p','no:cacheprovider','/work/tests/test_fight_r5.py','/work/tests/test_fight_r5_site.py','/work/tests/test_fight.py','/work/tests/test_loitering_settings.py','/work/tests/test_event_cooldown.py'],plugins=[r]) from pathlib import Path Path('/work/isolated-tests.json').write_text(json.dumps(dict(exit_code=int(code),collected=r.collected,passed=len(r.passed),failed=r.failed,errors=r.errors),indent=2)) raise SystemExit(code) """ out=self.cpu_job(name,built['image'],['-c',code],180) self.fp.atomic(self.root/'isolated-tests.log',out) print(json.dumps(load(self.root/'isolated-tests.json'))) def activate(self): tests=load(self.root/'isolated-tests.json') if tests['exit_code'] or tests['failed'] or tests['errors']: raise ValueError('R5_ISOLATED_TESTS_REQUIRED') super().activate() def migrate_settings(self): before=self.root/'settings-before';before.mkdir(mode=0o700) entries=[] for item in load(self.root/'config-targets.json'): path=self.s.safe(item['path']);info=path.stat() shutil.copy2(path,before/item['name']);(before/item['name']).chmod(0o600) entries.append(dict(item,before=sha(path),mode=stat.S_IMODE(info.st_mode),uid=info.st_uid,gid=info.st_gid)) self.save('settings-inputs',entries) bindings=self.state/'protected-bindings.json' shutil.copy2(bindings,before/'protected-bindings.json') built=load(self.root/'built.json') output=self.cpu_job('ct-r5-settings-20261101-r01',built['image'],['/work/fight_r5_settings.py','/work'],60) self.fp.atomic(self.root/'settings-plan.log',output) changes=load(self.root/'settings-migration.json') old_bindings=load(bindings);new_bindings=updated_bindings(old_bindings,changes) self.save('settings-commit',dict(phase='VALIDATED',completed=[],bindings_before=sha(bindings))) completed=[] for entry in changes: path=Path(entry['path']);raw=(self.root/'settings-after'/entry['name']).read_bytes() if sha(path)!=entry['before'] or sha(self.root/'settings-after'/entry['name'])!=entry['after']: raise ValueError('R5_SETTINGS_CAS_FAILED') if entry['changed']: self.fp.atomic(path,raw,mode=entry['mode']);os.chown(path,entry['uid'],entry['gid']) completed.append(entry['kind']) self.save('settings-commit',dict(phase='WRITING',completed=completed)) if load(bindings)!=old_bindings:raise ValueError('R5_BINDINGS_CHANGED_DURING_COMMIT') if new_bindings!=old_bindings:self.fp.atomic(bindings,new_bindings) self.save('settings-commit',dict(phase='COMMITTED',completed=completed,bindings_after=sha(bindings))) def finish_activation(self): plan,rows=self.guard() if not (self.root/'transition.json').exists() or rows['ai-gateway']['State']['Running']: raise ValueError('GATEWAY_HOLD_REQUIRED') roles=['analytics'] if plan['before']['analytics']['running'] else [] self.check_stopped(rows,roles) self.migrate_settings() active,_,_,approved,live,_=self.cb.inspect() if approved!=live:raise ValueError('COMPOSE_DRIFT') new=copy.deepcopy(approved);new['services']['analytics']['image']=load(self.root/'built.json')['image'] self.tx.mkdir(mode=0o700) for name,value in [('compose.json',new),('approved.json',new), ('compose.physical.json',self.s.physical(new)),('previous-active.json',active)]: self.fp.atomic(self.tx/name,value) from apply_source import normalize rendered=json.loads(command(self.cb.compose_args(self.tx/'compose.physical.json')+['config','--format','json'])) if normalize(self.s.logical(rendered))!=normalize(new):raise ValueError('COMPOSE_ROUNDTRIP') updated=dict(active,transaction=self.tx.name,compose=str(self.tx/'compose.json'), physical=str(self.tx/'compose.physical.json'),compose_semantic_sha256=self.cb.canonical(new)) self.save('activation',dict(before=active,after=updated)) if self.cb.active()!=active:raise ValueError('ACTIVE_CAS_FAILED') self.fp.atomic(self.state/'active.json',updated);self.cb.inspect();self.cb.guard_images(new) command(self.cb.compose_args(updated['physical'])+['up','--no-start','--no-deps','--no-build','--pull','never','analytics'],90) if roles:command(self.cb.compose_args(updated['physical'])+['start','analytics'],120) self.save('deployed',dict(active=updated,startedRoles=roles,gatewayHeld=plan['before']['ai-gateway']['running'])) print('R5 Analytics/config replaced; Gateway held for readiness. No camera Start.') 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'] until=time.monotonic()+120 while True: current=json.loads(command(['docker','inspect',row['Id']]))[0] if current['State'].get('Health',{}).get('Status')=='healthy':break if not current['State']['Running'] or time.monotonic()>until:raise ValueError('R5_READINESS_FAILED') time.sleep(2) if row['Image']!=load(self.root/'built.json')['image']:raise ValueError('R5_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('R5_SETTINGS_READBACK_FAILED') hashes=json.loads(self.python(row['Id'], 'import json,hashlib\nfrom pathlib import Path\nnames='+repr(list(self.m['after']))+ '\nprint(json.dumps({n:hashlib.sha256((Path("/opt/contextra")/n).read_bytes()).hexdigest() for n in names}))')) if hashes!=self.m['after']:raise ValueError('R5_DEPLOYED_SOURCE_MISMATCH') runtime=self.runtime(row['Id']) code="""import json,os,urllib.request from pathlib import Path ids=IDS rows=[] for sid in ids: with urllib.request.urlopen('http://127.0.0.1:5000/api/v1/streams/'+sid+'/settings',timeout=4) as r:d=json.load(r) for label in ('configured','effective'): assert d[label]['fight']['engine']=='r5' assert d[label]['fight']['person_score']>=.25 and d[label]['fight']['joint_score']>=.1 rows.append(dict(id=sid,engine=d['effective']['fight']['engine'],enabled=d['effective']['fight']['enabled'],person=d['effective']['fight']['person_score'],joint=d['effective']['fight']['joint_score'])) template=json.loads(Path(os.environ['CONTEXTRA_CONTROL_TEMPLATE_FILE']).read_text()) assert template['fight']['engine']=='r5' and not template['fight']['enabled'] 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) expected={'RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke','Awiros_person_and_head_detection'} assert expected.issubset({x['name'] for x in models if x['state']=='READY'}) print(json.dumps(dict(settings=rows,template_r5=True,models_ready=True))) """.replace('IDS',repr([s['streamId'] for s in runtime['streams']])) config=json.loads(self.python(row['Id'],code,60)) self.save('verified-before-gateway',dict(sourceVerified=True,runtime=runtime,**config)) if deployed['gatewayHeld']:command(['docker','start',rows['ai-gateway']['Id']],30) self.save('verified',dict(status='USER_BROWSER_WAIT',sourceVerified=True,channel_count=len(config['settings']), template_r5=True,models_ready=True,gatewayRestored=deployed['gatewayHeld'],cameraStartCalls=0)) print(json.dumps(load(self.root/'verified.json'))) if __name__=='__main__': root,expected,action=sys.argv[1:] if action not in ('prepare','build','test','activate','verify'): raise ValueError('R5_ACTION_NOT_ALLOWED') getattr(FightR5Installer(root,expected),action)()