import copy,hashlib,json,os,shutil,subprocess,sys,tarfile,time,urllib.request
from pathlib import Path
os.umask(0o077)
ROOT=Path('/backup/contextra/deployment-inbox/icerti-provider-20260927-r01')
NEW=Path('/workspace/contextra/runtime');DATA=Path('/workspace/contextra/data');tools=NEW/'deploy-tools/provider-20260927-r01'
sys.path.insert(0,str(tools))
from deployment_kit import site,release,compose_snapshot
from deployment_kit.core import atomic_write,write_json,sha256
def cmd(args,timeout=60):return subprocess.check_output(args,text=True,stderr=subprocess.STDOUT,timeout=timeout)
def inspect(name):return json.loads(cmd(['docker','inspect',name]))[0]
def api(path):return json.load(urllib.request.urlopen('http://127.0.0.1:35000'+path,timeout=5))
assert (ROOT/'migration-prepared.json').exists()
for name in ('analytics','ai-control-sidecar','ai-gateway','postgresql','redis'):
    assert not inspect('contextra-icerti-'+name+'-1')['State']['Running']
archive=Path('/home/si/icerti-provider-reserve-20260927-r01.tar.gz')
assert sha256(archive)=='fca4e57b5975154d12b74dfa38d2e251702528ca8bb13f0bf21dcc1be23f91f6'
shutil.copyfile(archive,ROOT/archive.name)
with tarfile.open(archive) as tar:tar.extractall(ROOT,filter='data')
update=ROOT/'reserve-update';hashes=json.loads((update/'manifest.json').read_text())
for p,h in hashes.items():assert sha256(update/p)==h
images=json.loads((ROOT/'images.json').read_text())
assert json.loads(cmd(['docker','image','inspect',images['analytics']['tag']]))[0]['Id']==images['analytics']['id']
tag='contextra-analytics:icerti-provider-d900910-20260927-r02'
with (ROOT/'build-reserve.log').open('x') as log:
    r=subprocess.run(['docker','build','--pull=false','--network=none','--memory=2g','--cpu-period=100000','--cpu-quota=200000','-t',tag,str(update)],
        stdout=log,stderr=log,env=dict(os.environ,DOCKER_BUILDKIT='0'),timeout=180)
assert r.returncode==0,'RESERVE_IMAGE'
images['analytics']=dict(tag=tag,id=json.loads(cmd(['docker','image','inspect',tag]))[0]['Id'])
write_json(ROOT/'images-final.json',images,new=True)
newsite=site.load(NEW/'state/site.json')
doc=json.loads((NEW/'state/transactions/provider-20260927-r01/compose.json').read_text())
doc['services']['analytics']['image']=images['analytics']['id']
doc['services']['analytics']['environment'].update(CONTEXTRA_ARTIFACT_EXTERNAL_RESERVE_BYTES=str(4*96*1024**2),
    CONTEXTRA_TRAINING_EXTERNAL_RESERVE_BYTES=str(64*128*1024**2))
tx=NEW/'state/transactions/provider-20260927-r02';tx.mkdir(mode=0o700)
for value in doc['services'].values():value.setdefault('labels',{})[release.TRANSACTION_LABEL]=tx.name
compose_snapshot.seal(doc,tx/'compose.json',newsite['project'])
overlay=copy.deepcopy(doc);overlay.pop('name',None)
overlaypath=DATA/'site/provider-20260927-r02.compose.json';write_json(overlaypath,overlay,new=True)
newsite['compose_files'][1]=str(overlaypath)
args,newhashes=release.compose_args(newsite,NEW/'releases'/json.loads((ROOT/'active-before.json').read_text())['release_id'])
persistent=json.loads(cmd(['docker',*args,'config','--format','json']))
for v in persistent['services'].values():v.pop('build',None)
def comparable(value):
    result=copy.deepcopy(value)
    for v in result['services'].values():v['volumes']=sorted(v.get('volumes',[]),key=lambda m:m['target'])
    return result
assert comparable(doc)==comparable(persistent),'PERSISTENT_DRIFT'
ids={k:v['image'] for k,v in doc['services'].items()}
release.validate_rendered(doc,newsite,ids)
receipts=release.model_receipts(doc,newsite,{'gpu':[{'name':'NVIDIA RTX PRO 4000 Blackwell'}]},{'tensorrt_version':'10.16.0'})
write_json(ROOT/'model-receipts.json',receipts,new=True)
write_json(NEW/'state/site.json',newsite,expected=sha256(NEW/'state/site.json'))
unit=Path('/etc/systemd/system/contextra-runtime.service')
atomic_write(unit,unit.read_text().replace('/transactions/provider-20260927-r01/','/transactions/provider-20260927-r02/'),mode=0o644)
cmd(['systemctl','daemon-reload'])
cmd(['systemctl','enable','contextra-runtime.service'])
subprocess.run(['/usr/bin/python3','-m','deployment_kit.site_guard','--site',str(NEW/'state/site.json'),'--compose',str(tx/'compose.json'),'check'],
    env=dict(os.environ,PYTHONPATH=str(tools)),check=True,timeout=30)
write_json(NEW/'state/inflight.json',dict(status='APPLYING',stage='PROVIDER_PATCH',transaction=tx.name))
compose=['docker','compose','-p',newsite['project'],'-f',str(tx/'compose.json')]
def ready(name,timeout):
    end=time.monotonic()+timeout
    while time.monotonic()<end:
        c=inspect('contextra-icerti-'+name+'-1')
        if not c['State']['Running'] or c['State']['OOMKilled']:raise RuntimeError('SERVICE_EXIT_'+name)
        if c['State'].get('Health',{}).get('Status')=='healthy':return
        time.sleep(2)
    raise RuntimeError('SERVICE_READY_'+name)
for name in ('postgresql','redis','analytics','ai-control-sidecar'):
    cmd(compose+['up','-d','--no-deps','--no-build','--pull','never',name],180)
    ready(name,150 if name=='analytics' else 60)
    print(name+' READY',flush=True)
cmd(['systemctl','restart','contextra-redis-access.service'],30)
before={c['Name'].lstrip('/'):c for c in json.loads((ROOT/'containers-private.json').read_text())}
for name,dest in [('postgresql','/var/lib/postgresql/data'),('redis','/data')]:
    old=next(m['Name'] for m in before['contextra-icerti-'+name+'-1']['Mounts'] if m['Destination']==dest)
    new=next(m['Name'] for m in inspect('contextra-icerti-'+name+'-1')['Mounts'] if m['Destination']==dest)
    assert old==new,'VOLUME_CHANGED_'+name
rows=api('/api/v1/streams')['streams'];assert len(rows)==44 and all(r['state']=='stopped' and not r.get('child_pid') for r in rows)
artifact=api('/api/v1/metadata/status')['artifacts']
assert artifact['enabled'] and artifact['http_enabled'] and artifact['storage_cap_bytes']==4_000_000_000_000
code='''import json,urllib.request
out={}
for model in ('RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke','Awiros_person_and_head_detection'):
 c=json.load(urllib.request.urlopen('http://localhost:8000/v2/models/'+model+'/config',timeout=3))
 assert c['max_batch_size']==8 and c['version_policy']['specific']['versions']==[8]
 assert c['dynamic_batching']['preferred_batch_size']==[8] and all(g['count']==8 for g in c['instance_group'])
 out[model]=urllib.request.urlopen('http://localhost:8000/v2/models/'+model+'/ready',timeout=3).status
print(json.dumps(out))'''
model_ready=json.loads(cmd(['docker','exec','contextra-icerti-analytics-1','python3','-B','-c',code],20))
token=(DATA/'site/bridge/control-token').read_text().strip()
def training(method='GET',body=None):
    r=urllib.request.Request('http://127.0.0.1:35000/internal/secure-control/training/settings',method=method,
        data=json.dumps(body).encode() if body else None,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'})
    return json.load(urllib.request.urlopen(r,timeout=5))
t=training();t=training('PATCH',dict(expected_revision=t['revision'],changes=dict(max_bytes=2_000_000_000_000,min_free_bytes=100_000_000_000)))
assert t['configured']['enabled'] and t['configured']['minimum_persons']==8 and t['configured']['interval_seconds']==900
assert t['configured']['width']==1280 and t['configured']['height']==720
write_json(ROOT/'training-site-settings.json',t,new=True)
# Disable only the approved whole-camera deletion job, preserving all other cron lines.
cron=cmd(['crontab','-l']);write_json(ROOT/'cron-before.json',dict(text=cron),new=True)
matches=[line for line in cron.splitlines() if line.strip() and not line.lstrip().startswith('#') and '/root/data_cleanup.sh' in line]
assert matches==['0 * * * * /usr/bin/flock -n /tmp/cameras-cleanup.lock /root/data_cleanup.sh'],'CRON_DRIFT'
newcron='\n'.join('# Disabled by ICERTI-STORAGE-01: Event ledger-owned retention replaces this job.' if line==matches[0] else line for line in cron.splitlines())+'\n'
subprocess.run(['crontab','-'],input=newcron,text=True,check=True,timeout=10)
assert cmd(['crontab','-l'])==newcron
active=json.loads((ROOT/'active-before.json').read_text())
active.update(transaction=tx.name,images=ids,compose_sha256=sha256(tx/'compose.json'),normalized_sha256=sha256(tx/'compose.normalized.json'),
    site_sha256=sha256(NEW/'state/site.json'),config_inputs=newhashes,model_receipts=receipts,
    bound_files=compose_snapshot.file_bindings(doc),time=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),
    incremental_patch=dict(id='icerti-provider-20260927-r01',source_head='d900910d04d2bcdc492a67c788ad6776fe9380de',
        manifest=str(ROOT/'kit/manifest.json'),manifest_sha256=sha256(ROOT/'kit/manifest.json'),
        update_manifest=str(update/'manifest.json'),update_manifest_sha256=sha256(update/'manifest.json'),tools=str(tools),
        note='Provider patch and source hashes supplement the preserved FULL baseline; canary pending'),
    acceptance=dict(DEPLOYED='PASS',MODELS='PASS',PATHS='PASS',CANARY='PENDING',WSS='GATEWAY_STOPPED'))
write_json(tx/'applied.json',active,new=True);write_json(NEW/'state/active.json',active)
write_json(NEW/'state/inflight.json',dict(status='DONE',stage='APPLIED_CANARY_PENDING',transaction=tx.name))
write_json(ROOT/'reflected.json',dict(image=images,models=model_ready,channels_stopped=len(rows),artifact_cap=artifact['storage_cap_bytes'],
 training=t['configured'],cron_replaced=True,gateway='stopped_until_canary',compose=str(tx/'compose.json')),new=True)
print(json.dumps(dict(status='REFLECTED_CANARY_PENDING',models=model_ready,stopped=len(rows))))