import copy,csv,hashlib,json,os,shutil,subprocess,sys,time,uuid
from pathlib import Path
import yaml
os.umask(0o077)
R=Path(__file__).parent;R.chmod(0o700)
REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor');sys.path.insert(0,str(REPO))
from tools.stream_benchmark import load_config
from tools.stream_benchmark_runtime import Runtime,file_hash
from app.stream_config_store import StreamConfigStore
from app.runtime_state import write_yaml_atomic
from tools.test_run_manager import load_preset
CFG=Path('/workspace/contextra-runtime-config/test-run/event-artifact-next.yaml')
REG=Path('/data/contextra/runtime-state/streams_runtime.yaml')
MANIFEST=Path('/workspace/contextra-runtime-config/test-run/medium16-registry.json')
PRESETS=Path('/workspace/contextra-runtime-config/test-run/presets')
ACTIVE=Path('/workspace/contextra-runtime-config/metadata-active/compose.yaml')
ROOT=Path('/data/contextra/artifacts')
PRIOR=Path('/tmp/codex/ltb1021/2026-09-17_contextra-event-web-id_r01/final.json')
TARGET={'web-cam08-test':2,'web-cam01-test':1}
REASON='OPERATOR_HOLD_SERVER_ID_MIGRATION_AI01'
def save(name,value):
 f=R/name;f.write_text(json.dumps(value,ensure_ascii=False,indent=2));f.chmod(0o600)
def read(name):return json.loads((R/name).read_text())
def cmd(args,timeout=30,input=None):
 p=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout)
 if p.returncode:
  save('command-error-'+str(time.time_ns())+'.json',dict(command=args,returncode=p.returncode,stdout=p.stdout[-2000:],stderr=p.stderr[-3000:]))
  raise RuntimeError('Command failed; see private evidence')
 return p.stdout
def pg(sql):
 return cmd(['docker','exec','-i','contextra-postgresql-1','sh','-c','psql -X -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB" -At'],input=sql,timeout=15)
def ledger():
 return json.loads(pg("SELECT coalesce(json_agg(x),'[]') FROM (SELECT * FROM contextra_event_artifacts ORDER BY created_at) x;"))
def rt():return Runtime(load_config(CFG),R)
def gate():
 assert not (R/'plan.json').exists()
 runtime=rt();state=runtime.inspect();assert state['State']['Status']=='exited' and state['State']['Pid']==0
 assert state['Image']==runtime.config['expected_image']
 old=ledger();actual={x['event_id']:x for x in old}
 prior=json.loads(PRIOR.read_text());targets=[x for x in prior['events'] if x['delivery_state']=='PENDING' and x['attempts']==0]
 assert len(targets)==16 and sum(x['artifact_state']=='READY' for x in targets)==1 and sum(x['artifact_state']=='CAPTURE_READY' for x in targets)==15
 assert all(actual[x['event_id']]==x for x in targets)
 assert all(x['delivery_state'] in ['PENDING','FAILED','SENT','BLOCKED_TIME'] for x in old)
 registry=StreamConfigStore(str(REG)).read();assert len(registry['added'])==18 and all(not x.get('autostart') for x in registry['added'])
 assert len({x['source_id'] for x in registry['added']})==18
 configs={str(f):dict(hash=file_hash(f),value=yaml.safe_load(f.read_text())) for f in PRESETS.rglob('*.yaml')}
 assert len(configs)==17
 assert shutil.disk_usage('/var/lib/docker').free>15*1024**3 and shutil.disk_usage(ROOT).free>60*1024**3
 files={}
 for row in old:
  for folder in [ROOT/row['relative_path'],ROOT/'.staging'/row['event_id']]:
   if folder.exists():
    files.update({str(f):file_hash(f) for f in folder.rglob('*') if f.is_file()})
 save('plan.json',dict(head=cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip(),at=time.time(),image=state['Image'],container_id=state['Id'],
   registry=registry,registry_hash=file_hash(REG),presets=configs,manifest=json.loads(MANIFEST.read_text()),manifest_hash=file_hash(MANIFEST),
   config=runtime.config,config_hash=file_hash(CFG),compose_hash=file_hash(ACTIVE),old_events=old,hold_targets=targets,artifact_hashes=files,
   redis_id=json.loads(cmd(['docker','inspect','contextra-redis-1']))[0]['Id'],postgres_id=json.loads(cmd(['docker','inspect','contextra-postgresql-1']))[0]['Id'],
   caps=dict(event_seconds_from_first_frame=60,control_start_seconds=90,container_ram_gib=16,free_disk_gib=15,gpu_mib=40000),failure='STOP_ONLY_NO_ROLLBACK'))
 print('Gate PASS: exact16 pending rows matched,18 stopped registrations, existing image',flush=True)
def hold():
 p=read('plan.json');assert not (R/'hold-intent.json').exists();assert not rt().inspect()['State']['Running']
 expected={e['event_id']:e for e in p['hold_targets']};actual={e['event_id']:e for e in ledger()};assert all(actual[k]==v for k,v in expected.items())
 for key in expected:uuid.UUID(key)
 save('hold-intent.json',dict(reason=REASON,before=list(expected.values())))
 payload=json.dumps(list(expected.values()),ensure_ascii=False).replace("'","''")
 sql="""BEGIN;
SET LOCAL statement_timeout='5s'; SET LOCAL lock_timeout='2s';
DO $hold$
DECLARE affected integer;
BEGIN
 WITH expected AS (SELECT value AS row FROM jsonb_array_elements('%s'::jsonb))
 UPDATE contextra_event_artifacts AS event
 SET delivery_state='HELD',delivery_error='%s',updated_at=extract(epoch from clock_timestamp())
 FROM expected WHERE event.event_id=expected.row->>'event_id' AND to_jsonb(event)=expected.row
 AND event.delivery_state='PENDING' AND event.attempts=0;
 GET DIAGNOSTICS affected=ROW_COUNT;
 IF affected<>16 THEN RAISE EXCEPTION 'Exact hold CAS mismatch'; END IF;
END $hold$;
COMMIT;
"""%(payload,REASON)
 pg(sql)
 after={x['event_id']:x for x in ledger()}
 for x in p['old_events']:
  y=copy.deepcopy(after[x['event_id']])
  if x['event_id'] in expected:
   assert y['delivery_state']=='HELD' and y['delivery_error']==REASON and y['attempts']==0
   for k in ['delivery_state','delivery_error','updated_at']:y[k]=x[k]
  assert y==x
 save('held.json',dict(rows=[after[k] for k in expected],reason=REASON,at=time.time()))
 print('Exact16 HELD persisted; payload/body/attempts/files preserved',flush=True)
def configure():
 p=read('plan.json');assert (R/'held.json').exists() and not (R/'configure-intent.json').exists()
 assert not rt().inspect()['State']['Running'] and file_hash(REG)==p['registry_hash']
 save('configure-intent.json',dict(mapping='all18 server_id=AI-01',before_hash=p['registry_hash']))
 store=StreamConfigStore(str(REG))
 def change(value):
  assert value==p['registry']
  for row in value['added']:row['server_id']='AI-01'
 store.update(change)
 after=store.read()
 for before,new in zip(p['registry']['added'],after['added']):
  assert new['server_id']=='AI-01';check=dict(new);check['server_id']=before['server_id'];assert check==before
 for name,entry in p['presets'].items():
  f=Path(name);assert file_hash(f)==entry['hash'];value=copy.deepcopy(entry['value']);value['camera']['server_id']='AI-01'
  write_yaml_atomic(str(f),value);f.chmod(0o600);assert load_preset(f)['camera']['server_id']=='AI-01'
 manifest=copy.deepcopy(p['manifest']);manifest['server_id']='AI-01'
 for row in manifest['channels']:row['server_id']='AI-01'
 assert file_hash(MANIFEST)==p['manifest_hash'];MANIFEST.write_text(json.dumps(manifest,ensure_ascii=False,indent=2));MANIFEST.chmod(0o600)
 for file in [CFG,CFG.with_name('event-artifact.yaml')]:
  value=load_config(file)
  for artifact in value['artifacts']:
   if artifact['path']==str(REG):artifact['sha256']=file_hash(REG)
  write_yaml_atomic(str(file),value);file.chmod(0o600)
 save('configured.json',dict(registry_hash=file_hash(REG),manifest_hash=file_hash(MANIFEST),registry=after,
   preset_hashes={name:file_hash(name) for name in p['presets']}))
 print('Permanent AI-01 applied atomically to18 registrations,17 presets, manifest/config hashes',flush=True)
def startup():
 p=read('plan.json');assert (R/'configured.json').exists() and not (R/'startup-intent.json').exists()
 assert file_hash(ACTIVE)==p['compose_hash'];runtime=rt();assert runtime.inspect()['Id']==p['container_id'] and not runtime.inspect()['State']['Running']
 save('startup-intent.json',dict(container_id=p['container_id'],at=time.time()))
 cmd(['docker','start','contextra-analytics-1'],30)
 deadline=time.monotonic()+120
 while time.monotonic()<deadline:
  try:
   states=runtime.states()
   if states:break
  except Exception:pass
  time.sleep(1)
 else:raise RuntimeError('API readiness timeout')
 assert all(x['state']=='stopped' and not x.get('child_pid') for x in states.values())
 now={e['event_id']:e for e in ledger()}
 for e in read('held.json')['rows']:assert now[e['event_id']]==e
 save('started.json',dict(states=states,model=runtime.triton('v2/models/RFDETRMedium/config'),resource=runtime.resources(),held_unchanged=True))
 print('Analytics ready; held16 unchanged, no channel started',flush=True)
def fresh():
 old={e['event_id'] for e in read('plan.json')['old_events']};return [e for e in ledger() if e['event_id'] not in old]
def cleanup():
 runtime=rt();result={}
 try:
  states=runtime.states()
  for sid in TARGET:
   if states[sid].get('child_pid') or states[sid]['state'] in ['running','starting']:
    try:result[sid]=runtime.stop(sid)
    except Exception as e:result[sid]=dict(error=type(e).__name__,actual=runtime.states().get(sid))
  save('cleanup.json',dict(results=result,states=runtime.states()))
 finally:
  if runtime.inspect()['State']['Running'] and not (R/'stop-container-intent.json').exists():
   save('stop-container-intent.json',dict(at=time.time()));cmd(['docker','stop','-t','80','contextra-analytics-1'],95)
def trial():
 runtime=rt();assert not (R/'trial-intent.json').exists();save('trial-intent.json',dict(at=time.time(),target=TARGET))
 failure=None;first_at=None;observations=[]
 try:
  for sid in TARGET:
   value=runtime.start(sid);save('first-frame-'+sid+'.json',value)
   if first_at is None:first_at=time.monotonic()
  deadline=first_at+60
  while time.monotonic()<deadline:
   entries=fresh();states=runtime.states();resource=runtime.resources()
   observations.append(dict(at=time.monotonic(),events=entries,states=states,resource=resource));save('observations.json',observations)
   for row in entries:
    assert row['payload']['serverId']=='AI-01' and row['payload']['cameraId'] in TARGET.values()
    if row['delivery_state'] in ['FAILED','BLOCKED_TIME'] or row['artifact_state']=='FAILED':raise RuntimeError('Event boundary failed')
   if any(states[sid]['state']!='running' or states[sid].get('last_error') for sid in TARGET):raise RuntimeError('Runtime channel failed')
   if all(any(e['payload']['cameraId']==i and e['artifact_state']=='READY' and e['delivery_state']=='SENT' for e in entries) for i in TARGET.values()):break
   time.sleep(1)
  else:raise RuntimeError('No confirmed Web acceptance before natural Event deadline')
  save('accepted.json',dict(elapsed=time.monotonic()-first_at,events=entries,states=states))
 except Exception as e:
  failure=str(e);save('failure.json',dict(error=failure,events=fresh(),states=runtime.states()));cleanup()
 finally:
  save('result.json',dict(failure=failure,elapsed=time.monotonic()-first_at if first_at else None,events=fresh(),
    states=runtime.states() if runtime.inspect()['State']['Running'] else read('cleanup.json')['states'],
    image=runtime.inspect()['Image'],container_state=runtime.inspect()['State']['Status']))
 print('Web acceptance: '+str(failure or 'SUCCESS for both cameras; two running'),flush=True)
def audit():
 runtime=rt();p=read('plan.json');now={e['event_id']:e for e in ledger()};held={e['event_id']:e for e in read('held.json')['rows']}
 for e in p['old_events']:assert now[e['event_id']]==held.get(e['event_id'],e)
 for name,h in p['artifact_hashes'].items():assert file_hash(name)==h
 data=StreamConfigStore(str(REG)).read();assert len(data['added'])==18 and all(e['server_id']=='AI-01' for e in data['added'])
 for name in p['presets']:assert load_preset(name)['camera']['server_id']=='AI-01'
 states=runtime.states() if runtime.inspect()['State']['Running'] else read('cleanup.json')['states']
 assert all(v['state']=='stopped' and not v.get('child_pid') for sid,v in states.items() if sid not in TARGET)
 assert runtime.inspect()['Image']==p['image']==load_config(CFG)['expected_image']
 for name,k in [('contextra-redis-1','redis_id'),('contextra-postgresql-1','postgres_id')]:assert json.loads(cmd(['docker','inspect',name]))[0]['Id']==p[k]
 sender=json.loads(cmd(['python3',str(REPO/'tools/rtsp_test.py'),'list']));assert all(x['state']=='STOPPED' for x in sender if x['name']=='sample01')
 rows=[]
 for sid,i in TARGET.items():
  entries=[e for e in fresh() if e['payload']['cameraId']==i]
  rows.append(dict(stream_id=sid,camera_id=i,server_id='AI-01',state=states[sid]['state'],events=len(entries),
    ready=sum(e['artifact_state']=='READY' for e in entries),accepted=sum(e['delivery_state']=='SENT' for e in entries),
    attempts=sum(e['attempts'] for e in entries),failed=sum(e['delivery_state']=='FAILED' for e in entries)))
 save('summary.json',dict(rows=rows,held_unchanged=16,old_rows_and_files_preserved=True,web_result=read('result.json')['failure'] or 'SUCCESS',
   runtime_result='STOPPED_PENDING_CAP' if (R/'capacity.json').exists() else 'RUNNING',
   elapsed=read('result.json')['elapsed'],image=p['image'],sender=sender,states=states,events=fresh()))
 with (R/'summary.csv').open('w') as f:
  w=csv.DictWriter(f,fieldnames=list(rows[0]));w.writeheader();w.writerows(rows)
 print('Final audit PASS: held16 untouched, AI-01 persistent, old files/rows and infrastructure preserved',flush=True)
def capacity_record():
 save('capacity.json',dict(observed_after_web_success=True,at=time.time(),first_observed_counts=dict(capture_ready=47,capturing=1,rendering=1,ready_sent=25,failed_pending_cap=33),events=fresh()))
 print('Capacity failure recorded separately from Web success',flush=True)
if __name__=='__main__':globals()[sys.argv[1]]()