"""TRACE-03: exact-baseline local deployment; no Start, enrollment or model mutation."""
import argparse
import copy
import hashlib
import json
import os
from pathlib import Path
import pwd
import shutil
import subprocess
import sys
import time


def command(args,timeout=30,input=None):
    p=subprocess.run(args,input=input,capture_output=True,timeout=timeout)
    if p.returncode:raise RuntimeError('TRACE_COMMAND_FAILED '+str(args[:3])+' '+p.stderr.decode(errors='replace')[-500:])
    return p.stdout


def sha(p):return hashlib.sha256(Path(p).read_bytes()).hexdigest()


def load(p):return json.loads(Path(p).read_text())


def image_payload(source,destination,names):
    """Code only: host umask/ACL must not make non-root image imports unreadable."""
    source,destination=Path(source),Path(destination)
    destination.mkdir(parents=True,exist_ok=True);destination.chmod(0o755)
    for name in names:
        relative=Path(name)
        if relative.is_absolute() or '..' in relative.parts:raise ValueError('TRACE_PAYLOAD_PATH')
        path=destination/relative
        path.parent.mkdir(parents=True,exist_ok=True)
        for parent in path.parents:
            if parent==destination:break
            if parent.is_symlink():raise ValueError('TRACE_PAYLOAD_SYMLINK')
            parent.chmod(0o755)
        if path.is_symlink() or (source/relative).is_symlink():raise ValueError('TRACE_PAYLOAD_SYMLINK')
        shutil.copyfile(source/relative,path);path.chmod(0o644)


def export_directory(row):
    env=dict(x.split('=',1) for x in row['Config']['Env'] if '=' in x)
    target=env.get('CONTEXTRA_ARTIFACT_ROOT','/data/contextra/artifacts')
    matches=[m['Source'] for m in row['Mounts'] if m['Type']=='bind' and m['Destination']==target]
    if len(matches)!=1:raise ValueError('TRACE_ARTIFACT_MOUNT_REQUIRED')
    return Path(matches[0]).parent/'diagnostics/exports'


def policy_pair(src,applied,settings):
    """Retained, identical staging policies may be reused, never silently replaced."""
    if src.is_symlink() or applied.is_symlink():raise ValueError('TRACE_POLICY_SYMLINK')
    exists=(src.exists(),applied.exists())
    if not any(exists):return False
    if not all(exists):raise ValueError('TRACE_POLICY_PAIR_INCOMPLETE')
    if load(src)!=settings or load(applied)!=settings:raise ValueError('TRACE_POLICY_CHANGED_REVIEW_REQUIRED')
    return True


class Installer:
    def __init__(self,root,expected):
        if os.geteuid()!=0:raise ValueError('ROOT_REQUIRED')
        self.root=Path(root)
        self.definition=load('/etc/contextra-maintenance/server.json')
        if self.definition['server']!=expected:raise ValueError('TRACE_SERVER_MISMATCH')
        sys.path.insert(0,self.definition['paths']['tools'])
        import server_context as server
        import field_profiles as fp
        import compose_backend as cb
        self.s,self.fp,self.cb=server,fp,cb
        server.safe(self.root);self.root.chmod(0o700)
        self.m=load(self.root/'manifest.json')
        if self.m['server']!=expected:raise ValueError('TRACE_PAYLOAD_SERVER')
        self.state=cb.state_path()
        self.tx=self.state/'transactions'/('trace-'+self.root.name)
        self.tools=Path(self.definition['paths']['tools'])
        self.guard_payload()
        self.smoke=self.root/'trace_candidate_smoke.py'

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

    def guard_payload(self):
        for name,h in self.m['after'].items():
            p=self.s.safe(self.root/'payload'/name)
            if not p.is_relative_to(self.root/'payload') or sha(p)!=h:raise ValueError('TRACE_PAYLOAD_CHANGED')
        if shutil.disk_usage('/').free<15*1024**3:raise ValueError('ROOT_RESERVE_15GIB')
        self.s.guard_disks()

    def rows(self):return {r['Config']['Labels']['com.docker.compose.service']:r for r in self.cb.containers()}

    def python(self,container,code,timeout=15):
        return command(['docker','exec','-i',container,'python3','-B','-'],timeout,code.encode())

    def runtime(self,container):
        return json.loads(self.python(container,"""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:print(r.read().decode())
"""))

    def drain_events(self,container):
        deadline=time.monotonic()+75
        while True:
            n=json.loads(self.python(container,"""import json,os,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 ('CLAIMED','SENDING')")
  print(json.dumps(q.fetchone()[0]))
"""))
            if n==0:return
            if time.monotonic()>deadline:raise ValueError('TRACE_EVENT_WORK_NOT_DRAINED')
            time.sleep(2)

    def session(self,container):
        result=json.loads(self.python(container,"""import json,time,hashlib
from pathlib import Path
from control_sidecar.config import Config
from control_sidecar.auth import Auth
from control_sidecar.protocol import TOKEN
c=Config.from_env();a=Auth(c,None);p=Path(c.session_file)
raw=p.read_bytes();d=json.loads(raw);expiry=d.get('expires')
ok=(not p.is_symlink() and not p.stat().st_mode&0o077 and d.get('identity')==a._session_identity()
    and bool(TOKEN.fullmatch(d.get('token',''))) and (expiry is None or expiry>time.time()*1000+300000))
print(json.dumps({'validForRestart':bool(ok),'sha256':hashlib.sha256(raw).hexdigest(),'mtime':p.stat().st_mtime}))
"""))
        if not result['validForRestart']:raise ValueError('TRACE_SESSION_RESTART_REQUIRES_REVIEW')
        return result

    def prepare(self):
        if (self.root/'plan.json').exists():raise ValueError('PRESERVE_EXISTING_PLAN')
        self.s.verify_tools()
        for name in ('pending.json','apply-pending.json','network-pending.json'):
            if (self.fp.STATE/name).exists():raise ValueError('TRACE_PENDING_OPERATION_PRESERVED '+name)
        a,site,cp,approved,live,source=self.cb.inspect()
        if approved!=live:raise ValueError('REVIEW_CONFIG_DRIFT')
        rows=self.rows()
        if set(rows)!=set(self.definition['services'].values()):raise ValueError('TARGET_SERVICES_CHANGED')
        for role,r in rows.items():
            if r['Image']!=self.m['images'][role]:raise ValueError('TRACE_BASE_IMAGE_CHANGED')
            cfg=approved['services'][role]
            if cfg['image'] not in (r['Image'],r['Config']['Image']):raise ValueError('TRACE_IMAGE_CONFIG_DRIFT '+role)
            env=dict(x.split('=',1) for x in r['Config']['Env'] if '=' in x)
            if any(str(v)!=env.get(k) for k,v in cfg.get('environment',{}).items()):
                raise ValueError('TRACE_ENV_CONFIG_DRIFT '+role)
            actual={m['Destination']:m for m in r['Mounts']}
            for mount in cfg.get('volumes',[]):
                m=actual.get(mount['target'])
                if not m or m['RW']==bool(mount.get('read_only',False)):
                    raise ValueError('TRACE_MOUNT_CONFIG_DRIFT '+role)
                if mount['type']=='bind' and mount['source']!=m['Source']:
                    raise ValueError('TRACE_BIND_CONFIG_DRIFT '+role)
        for name,h in self.m['before'].items():
            if name=='ct.py':
                if sha(self.tools/name)!=h:raise ValueError('TRACE_CT_CHANGED')
            else:
                role='analytics' if name.startswith('app/') else 'ai-control-sidecar'
                base='/opt/contextra/' if role=='analytics' else '/opt/bridge/'
                dest=self.root/'baseline'/name;dest.parent.mkdir(parents=True,exist_ok=True)
                command(['docker','cp',rows[role]['Id']+':'+base+name,str(dest)])
                if sha(dest)!=h:raise ValueError('TRACE_RUNNING_SOURCE_CHANGED '+name)
        backup=self.root/'tool-backup';backup.mkdir(mode=0o700)
        names=['installed-files.json',*[n for n in self.m['after'] if not n.startswith(('app/','control_sidecar/'))]]
        existed=[]
        for name in names:
            p=self.tools/name
            if p.exists():
                target=backup/name;target.parent.mkdir(parents=True,exist_ok=True);shutil.copy2(p,target);existed.append(name)
        protected={str(p):sha(p) for p in (self.state/'site.json',self.state/'protected-bindings.json',self.state/'input-baseline.json') if p.exists()}
        before={role:{'id':r['Id'],'image':r['Image'],'running':r['State']['Running'],
                     'restart':r['HostConfig']['RestartPolicy'],'mounts':r['Mounts']} for role,r in rows.items()}
        session=self.session(rows['ai-control-sidecar']['Id']) if rows['ai-control-sidecar']['State']['Running'] else None
        self.save('plan',dict(active=a,activeHash=sha(self.state/'active.json'),sourceHash=self.cb.canonical(source),
                            protected=protected,before=before,session=session,toolInventoryHash=sha(self.tools/'installed-files.json'),toolExisted=existed))
        print(json.dumps({'phase':'PREPARED','server':self.m['server'],'runtime':self.runtime(rows['analytics']['Id'])['streamCounts'] if rows['analytics']['State']['Running'] else 'stopped'}))

    def guard(self,tools=True):
        self.guard_payload();p=load(self.root/'plan.json')
        if self.cb.active()!=p['active'] or sha(self.state/'active.json')!=p['activeHash']:
            raise ValueError('TRACE_ACTIVE_CHANGED')
        self.cb.inspect()
        if self.cb.canonical(self.cb.render())!=p['sourceHash']:raise ValueError('TRACE_SOURCE_CONFIG_CHANGED')
        if any(sha(path)!=h for path,h in p['protected'].items()):raise ValueError('TRACE_PROTECTED_CHANGED')
        if tools and sha(self.tools/'installed-files.json')!=p['toolInventoryHash']:raise ValueError('TRACE_TOOLS_CHANGED')
        rows=self.rows()
        if any(rows[k]['Id']!=v['id'] or rows[k]['Image']!=v['image'] for k,v in p['before'].items()):
            raise ValueError('TRACE_CONTAINERS_CHANGED')
        return p,rows

    def build(self):
        p,_=self.guard();result={}
        for role,basepath in [('analytics','/opt/contextra'),('ai-control-sidecar','/opt/bridge')]:
            saved=self.root/('built-'+role+'.json')
            if saved.exists():result[role]=load(saved);continue
            ctx=self.root/('build-'+role);ctx.mkdir(exist_ok=True,mode=0o700)
            prefix='app/' if role=='analytics' else 'control_sidecar/'
            paths=[n for n in self.m['after'] if n.startswith((prefix,'contextra_trace/'))]
            image_payload(self.root/'payload',ctx/'payload',paths)
            base=p['before'][role]['image']
            lines=['FROM '+base,'COPY payload/ '+basepath+'/']
            modules='app.trace_runtime,app.trace_events,app.secure_control' if role=='analytics' else 'control_sidecar.main'
            # Existing hotfix chains can be near overlayfs' layer bound. Verify in a
            # disposable container rather than committing an otherwise empty RUN layer.
            self.fp.atomic(ctx/'Dockerfile',('\n'.join(lines)+'\n').encode())
            tag='contextra-trace:'+self.m['server']+'-'+role+'-'+self.root.name
            with (self.root/('build-'+role+'.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(ctx)],
                    stdout=log,stderr=subprocess.STDOUT,timeout=300,env=dict(os.environ,DOCKER_BUILDKIT='0'))
            if r.returncode:raise RuntimeError('TRACE_BUILD_FAILED '+role)
            image=json.loads(command(['docker','image','inspect',tag]))[0]
            old=json.loads(command(['docker','image','inspect',base]))[0]
            if image['Config']!=old['Config']:raise ValueError('IMAGE_EXECUTION_CONFIG_CHANGED')
            if image['RootFS']['Layers'][:len(old['RootFS']['Layers'])]!=old['RootFS']['Layers']:raise ValueError('IMAGE_BASE_CHANGED')
            name='ct-trace-import-'+self.m['server']+'-'+role
            try:
                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','import '+modules],30)
            finally:
                leftover=subprocess.run(['docker','inspect',name],capture_output=True,timeout=10)
                if leftover.returncode==0:
                    check=json.loads(leftover.stdout)[0]
                    if check['Image']!=image['Id']:raise ValueError('TRACE_HELPER_IDENTITY')
                    command(['docker','rm','-f',name],15)
            if self.smoke.exists():
                smoke_name=name+'-lifespan'
                try:
                    output=command(['docker','run','--rm','--pull=never','-i','--name',smoke_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'],'-B','-',role],30,self.smoke.read_bytes())
                    self.fp.atomic(self.root/('smoke-'+role+'.json'),json.loads(output))
                finally:
                    leftover=subprocess.run(['docker','inspect',smoke_name],capture_output=True,timeout=10)
                    if leftover.returncode==0:
                        check=json.loads(leftover.stdout)[0]
                        if check['Image']!=image['Id']:raise ValueError('TRACE_HELPER_IDENTITY')
                        command(['docker','rm','-f',smoke_name],15)
            result[role]={'image':image['Id'],'tag':tag,'base':base}
            self.fp.atomic(saved,result[role]);print(json.dumps({'built':role,'image':image['Id']}),flush=True)
        self.save('built',result)

    def install_tools(self):
        self.guard()
        sys.path.insert(0,str(self.root/'payload'))
        from contextra_trace.writer import DEFAULT
        settings=dict(DEFAULT,min_free_bytes=15*1024**3)
        src=Path(self.definition['paths']['setup'])/'trace/trace-policy.json'
        applied=self.state/'trace/trace-policy.applied.json'
        self.s.safe(src);self.s.safe(applied)
        retained=policy_pair(src,applied,settings)
        for name in self.m['after']:
            if name.startswith(('app/','control_sidecar/')):continue
            path=self.tools/name
            self.fp.atomic(path,(self.root/'payload'/name).read_bytes(),mode=0o644)
        from tool_permissions import tool_access
        tool_access(self.tools)
        inventory=load(self.tools/'installed-files.json')
        inventory.update({n:h for n,h in self.m['after'].items() if not n.startswith(('app/','control_sidecar/'))})
        self.fp.atomic(self.tools/'installed-files.json',inventory,mode=0o644);self.s.verify_tools()
        for path in (src,applied):
            path.parent.mkdir(mode=0o755,parents=True,exist_ok=True);path.parent.chmod(0o755)
            if not retained:self.fp.atomic(path,settings,mode=0o644)
        user=pwd.getpwnam(self.definition['operator'])
        os.chown(src.parent,user.pw_uid,user.pw_gid);os.chown(src,user.pw_uid,user.pw_gid)
        export=export_directory(self.rows()['analytics']);data=export.parent.parent
        self.s.safe(export)
        if not data.is_dir() or data.is_symlink():raise ValueError('TRACE_DATA_ROOT_REQUIRED')
        if not export.parent.exists():
            export.parent.mkdir(mode=0o755);export.parent.chmod(0o755)
        export.mkdir(mode=0o700,exist_ok=True)
        os.chown(export,user.pw_uid,user.pw_gid)
        self.fp.atomic(self.fp.STATE/'trace-installation.json',dict(server=self.m['server'],policy=str(applied),exports=str(export),device=export.stat().st_dev))
        self.save('tools-installed',dict(policy=str(applied),editable=str(src),exports=str(export),inventoryHash=sha(self.tools/'installed-files.json'),retainedPolicies=retained))

    def failure_evidence(self,action):
        # Capture before a failed container can be replaced; never persist raw logs.
        sys.path.insert(0,str(self.root/'payload'))
        from contextra_trace.sanitize import copy_safe
        import traceback
        data={'action':action,'failure':traceback.format_exc()[-16000:],'services':{}}
        for role,row in self.rows().items():
            if role not in ('analytics','ai-control-sidecar','ai-gateway'):continue
            p=subprocess.run(['docker','logs','--tail','160',row['Id']],capture_output=True,timeout=12)
            data['services'][role]={'id':row['Id'],'image':row['Image'],'state':row['State'],
                'logTail':(p.stdout+p.stderr).decode(errors='replace')[-48000:]}
        clean,_=copy_safe(data)
        self.save('failure-'+action+'-'+str(time.time_ns()),clean)

    def activate(self):
        p,rows=self.guard(tools=False);built=load(self.root/'built.json');meta=load(self.root/'tools-installed.json')
        if sha(self.tools/'installed-files.json')!=meta['inventoryHash']:raise ValueError('TRACE_TOOL_DRIFT')
        gateway=rows['ai-gateway']
        if rows['ai-control-sidecar']['State']['Running']:
            self.session(rows['ai-control-sidecar']['Id'])
        if gateway['State']['Running']:
            command(['docker','stop','--signal','SIGTERM','--timeout','30',gateway['Id']],40)
        self.save('transition',{'phase':'GATEWAY_HELD','cameraStart':False})
        if rows['analytics']['State']['Running']:
            stop=(self.tools/'stop_channels.py').read_bytes()
            out=command(['docker','exec','-i',rows['analytics']['Id'],'python3','-B','-'],240,stop)
            self.fp.atomic(self.root/'stop-channels.log',out)
            status=self.runtime(rows['analytics']['Id'])
            if any(status['streamCounts'][k] for k in ('running','starting','error')):raise ValueError('EXTERNAL_START_CONFLICT')
            self.drain_events(rows['analytics']['Id'])
        roles=[k for k in ('analytics','ai-control-sidecar') if rows[k]['State']['Running']]
        if roles:command(self.cb.compose_args(p['active']['physical'])+['stop','--timeout','90',*roles],120)
        self.finish_activation()

    def check_stopped(self,rows,roles):
        evidence={}
        for role in roles:
            state=rows[role]['State'];okay=not state['Running'] and not state.get('OOMKilled') and state['ExitCode']==0
            markers={}
            # Uvicorn may re-raise SIGTERM after completing its lifespan shutdown.
            # Accept 143 only for this service and this transition's shutdown markers.
            if role=='ai-control-sidecar' and not state['Running'] and not state.get('OOMKilled') and state['ExitCode']==143:
                p=subprocess.run(['docker','logs','--since',str(int((self.root/'transition.json').stat().st_mtime)),rows[role]['Id']],capture_output=True,timeout=10)
                logs=(p.stdout+p.stderr).decode(errors='replace')
                markers={k:k in logs for k in ('Shutting down','Application shutdown complete','Finished server process')}
                okay=p.returncode==0 and all(markers.values())
            evidence[role]=dict(exit=state['ExitCode'],oom=state.get('OOMKilled',False),shutdownMarkers=markers,confirmed=okay)
            if not okay:
                self.save('stop-evidence',evidence)
                raise ValueError('TRACE_SERVICE_STOP_INCOMPLETE '+role)
        self.save('stop-evidence',evidence)

    def finish_activation(self):
        if not (self.root/'transition.json').exists():raise ValueError('TRACE_TRANSITION_REQUIRED')
        p,rows=self.guard(tools=False);built=load(self.root/'built.json');meta=load(self.root/'tools-installed.json')
        if sha(self.tools/'installed-files.json')!=meta['inventoryHash']:raise ValueError('TRACE_TOOL_DRIFT')
        if rows['ai-gateway']['State']['Running']:raise ValueError('TRACE_GATEWAY_NOT_HELD')
        roles=[k for k in ('analytics','ai-control-sidecar') if p['before'][k]['running']]
        self.check_stopped(rows,roles)
        a,_,_,approved,live,_=self.cb.inspect()
        if a!=p['active'] or approved!=live:raise ValueError('TRACE_ACTIVE_DRIFT')
        new=copy.deepcopy(approved)
        for role in ('analytics','ai-control-sidecar'):
            s=new['services'][role];s['image']=built[role]['image']
            s.setdefault('environment',{}).update(CONTEXTRA_TRACE_POLICY_FILE='/run/contextra-trace/trace-policy.applied.json',CONTEXTRA_TRACE_SERVER=self.m['server'])
            if any(v['target']=='/run/contextra-trace' for v in s.get('volumes',[])):raise ValueError('TRACE_BIND_EXISTS')
            s.setdefault('volumes',[]).append({'type':'bind','source':str(Path(meta['policy']).parent),
                'target':'/run/contextra-trace','read_only':True,'bind':{'create_host_path':False}})
        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',a)]:self.fp.atomic(self.tx/name,value)
        from apply_source import normalize
        reread=json.loads(command(self.cb.compose_args(self.tx/'compose.physical.json')+['config','--format','json']))
        if normalize(self.s.logical(reread))!=normalize(new):raise ValueError('TRACE_COMPOSE_ROUNDTRIP')
        updated=dict(a,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',{'before':a,'after':updated})
        if self.cb.active()!=a:raise ValueError('TRACE_ACTIVE_CAS')
        self.fp.atomic(self.state/'active.json',updated);self.cb.inspect();self.cb.guard_images(new)
        # Create stopped services with the new image, but never start a previously stopped service.
        command(self.cb.compose_args(updated['physical'])+['up','--no-start','--no-deps','--no-build','--pull','never','analytics','ai-control-sidecar'],90)
        if roles:command(self.cb.compose_args(updated['physical'])+['start',*roles],120)
        self.save('deployed',{'active':updated,'gatewayHeld':p['before']['ai-gateway']['running'],'startedRoles':roles})
        print('Target services replaced. Gateway held pending 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('TRACE_ACTIVE_CHANGED')
        rows=self.rows();result={}
        for role in deployed['startedRoles']:
            until=time.monotonic()+120
            while True:
                current=json.loads(command(['docker','inspect',rows[role]['Id']]))[0]
                if current['State'].get('Health',{}).get('Status')=='healthy':break
                if not current['State']['Running'] or time.monotonic()>until:raise ValueError('TRACE_READINESS_FAILED '+role)
                time.sleep(2)
        if 'analytics' in deployed['startedRoles']:
            out=command(['/usr/local/bin/ct','trace','status'],20).decode()
            self.fp.atomic(self.root/'ct-trace-status.txt',out.encode())
            if 'TRACE_DB_UNAVAILABLE' in out or 'TRACE_CONTROL_UNAVAILABLE' in out:raise ValueError('TRACE_READBACK_UNAVAILABLE')
            result['channelsBeforeGateway']=self.runtime(rows['analytics']['Id'])['streamCounts']
        for path,h in plan['protected'].items():
            if sha(path)!=h:raise ValueError('TRACE_PROTECTED_CHANGED')
        for role in ('redis','postgresql'):
            if rows[role]['Id']!=plan['before'][role]['id']:raise ValueError('UNRELATED_SERVICE_CHANGED')
        if plan.get('session'):
            if self.session(rows['ai-control-sidecar']['Id'])['sha256']!=plan['session']['sha256']:
                raise ValueError('TRACE_SESSION_CHANGED_REVIEW_REQUIRED')
        for role,base in [('analytics','/opt/contextra/'),('ai-control-sidecar','/opt/bridge/')]:
            prefix='app/' if role=='analytics' else 'control_sidecar/'
            expected={n:h for n,h in self.m['after'].items() if n.startswith((prefix,'contextra_trace/'))}
            actual=json.loads(self.python(rows[role]['Id'], 'import hashlib,json\nfrom pathlib import Path\nnames='+repr(list(expected))+'\nbase='+repr(base)+'\nprint(json.dumps({n:hashlib.sha256(Path(base+n).read_bytes()).hexdigest() for n in names}))'))
            if actual!=expected:raise ValueError('TRACE_INSTALLED_HASH_MISMATCH '+role)
            policy=json.loads(self.python(rows[role]['Id'],"import json,os\nfrom contextra_trace.writer import policy\nprint(json.dumps(policy(os.environ['CONTEXTRA_TRACE_POLICY_FILE'])))"))
            if not policy['enabled']:raise ValueError('TRACE_POLICY_DISABLED')
        result['payloadHashesVerified']=True
        if deployed['gatewayHeld']:
            command(['docker','start',rows['ai-gateway']['Id']],30)
        result['services']={k:{'running':r['State']['Running'],'image':r['Image']} for k,r in self.rows().items()}
        self.save('verified',result);print(json.dumps(result))

    def rollback(self):
        if (self.root/'rollback-started.json').exists():raise ValueError('TRACE_ROLLBACK_ALREADY_ATTEMPTED')
        p=load(self.root/'plan.json');activation=load(self.root/'activation.json')
        if self.cb.active()!=activation['after']:raise ValueError('TRACE_ROLLBACK_ACTIVE_DRIFT')
        self.cb.inspect();self.guard_payload()
        if any(sha(path)!=h for path,h in p['protected'].items()):raise ValueError('TRACE_ROLLBACK_PROTECTED_DRIFT')
        meta=load(self.root/'tools-installed.json')
        if sha(self.tools/'installed-files.json')!=meta['inventoryHash']:raise ValueError('TRACE_ROLLBACK_TOOLS_DRIFT')
        rows=self.rows()
        if rows['ai-gateway']['State']['Running']:raise ValueError('TRACE_ROLLBACK_REQUIRES_HELD_GATEWAY')
        if any(rows[k]['Id']!=p['before'][k]['id'] for k in ('redis','postgresql')):raise ValueError('TRACE_ROLLBACK_OTHER_SERVICE_DRIFT')
        self.failure_evidence('before-rollback')
        self.save('rollback-started',{'active':self.cb.active(),'restore':p['active']})
        roles=[k for k in ('analytics','ai-control-sidecar') if rows[k]['State']['Running']]
        if roles:command(self.cb.compose_args(activation['after']['physical'])+['stop','--timeout','90',*roles],120)
        self.check_stopped(self.rows(),roles)
        for name in p['toolExisted']:
            self.fp.atomic(self.tools/name,(self.root/'tool-backup'/name).read_bytes(),mode=0o644)
        self.s.verify_tools()
        self.fp.atomic(self.state/'active.json',p['active']);self.cb.inspect()
        command(self.cb.compose_args(p['active']['physical'])+['up','--no-start','--no-deps','--no-build','--pull','never','analytics','ai-control-sidecar'],90)
        restore=[k for k in ('analytics','ai-control-sidecar') if p['before'][k]['running']]
        if restore:command(self.cb.compose_args(p['active']['physical'])+['start',*restore],120)
        for role in restore:
            until=time.monotonic()+120
            while True:
                row=self.rows()[role]
                if row['Image']!=p['before'][role]['image']:raise ValueError('TRACE_ROLLBACK_IMAGE_DRIFT')
                if row['State'].get('Health',{}).get('Status')=='healthy':break
                if not row['State']['Running'] or time.monotonic()>until:raise ValueError('TRACE_ROLLBACK_NOT_READY')
                time.sleep(2)
        if p.get('session') and self.session(self.rows()['ai-control-sidecar']['Id'])['sha256']!=p['session']['sha256']:
            raise ValueError('TRACE_ROLLBACK_AUTH_CHANGED')
        if p['before']['ai-gateway']['running']:command(['docker','start',rows['ai-gateway']['Id']],30)
        self.save('rollback-complete',{'active':p['active'],'restoredRoles':restore,'diagnosticDataPreserved':True})
        print('Previous services restored once. No data/auth rollback or camera Start.')


def main():
    p=argparse.ArgumentParser();p.add_argument('action',choices=('prepare','build','install-tools','activate','finish-activation','verify','rollback'));p.add_argument('--root',required=True);p.add_argument('--expected-server',required=True)
    a=p.parse_args();i=Installer(a.root,a.expected_server)
    try:getattr(i,a.action.replace('-','_'))()
    except Exception:
        i.failure_evidence(a.action)
        raise SystemExit('TRACE_ACTION_FAILED; see masked failure evidence under '+str(i.root)) from None


if __name__=='__main__':main()