import importlib.util,json,os,pathlib,subprocess,urllib.request
os.umask(0o077);R=pathlib.Path(__file__).parent
sp=importlib.util.spec_from_file_location('gate','/tmp/codex/ltb1021/2026-09-11_service-stop_r01/gate.py')
g=importlib.util.module_from_spec(sp);sp.loader.exec_module(g)
AUTH=pathlib.Path('/tmp/codex/ltb1021/2026-09-11_contextra-metadata-auth_r01')
IMAGE='sha256:ca25695b7210d937442dfd572415a717c8caaab1de709f5f606f4f3e4c7bf3ed'
def main():
 assert not (R/'preflight.json').exists()
 assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()=='57605e940c31bcd19cc1d3759ee8faaa859b32d6'
 assert not g.sh(['git','-C',str(g.REPO),'status','--porcelain']).strip()
 assert json.loads(g.sh(['docker','image','inspect',IMAGE]))[0]['Id']==IMAGE
 snap=AUTH/'snapshot';files=list((g.REPO/'app').rglob('*.py'))
 assert all((snap/p.relative_to(g.REPO)).exists() and g.sha(p)==g.sha(snap/p.relative_to(g.REPO)) for p in files)
 assert g.sha(g.REPO/'docker/entrypoint.sh')==g.sha(snap/'docker/entrypoint.sh')
 c=g.inspect();assert c['State']['Status']=='running' and c['State']['Health']['Status']=='healthy'
 env=g.load_env_file(str(g.ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
 specs=g.yaml.safe_load(overlay.read_text())['added'];assert len(specs)==1
 assert all(specs[0][k] is False for k in ['store','active_learning','autostart'])
 opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
 with opener.open('http://127.0.0.1:'+env['FASTAPI_HOST_PORT']+'/api/v1/streams',timeout=3) as f:states=json.load(f)['streams']
 assert len(states)==1 and states[0]['state']=='stopped'
 redis=json.loads(g.sh(['docker','inspect','contextra-redis-1']))[0]
 def query(*args):return g.sh(['docker','exec',redis['Id'],'redis-cli','--raw',*args]).strip()
 assert query('DBSIZE')=='0' and not any(x.startswith('db') for x in query('INFO','keyspace').splitlines())
 assert 'nopass' in query('ACL','LIST')
 assert all(next((x[4:] for x in row.split() if x.startswith('cmd=')),None) in ('ping','client|list') for row in query('CLIENT','LIST').splitlines())
 oldplan=json.loads((AUTH/'plan.json').read_text());C=pathlib.Path(oldplan['controls'])
 assert all(g.sha(p)==h for p,h in oldplan['control_hashes'].items())
 override=g.yaml.safe_load((C/'compose.yaml').read_text());override['services']['redis'].pop('ports')
 override['services']['analytics']['image']=IMAGE
 (R/'diagnostic-compose.yaml').write_text(g.yaml.safe_dump(override))
 compose=['docker','compose','--env-file',str(g.ENV),'-f',str(g.REPO/'docker-compose.yml'),'-f',str(R/'diagnostic-compose.yaml')]
 rendered=json.loads(g.sh(compose+['config','--format','json']))
 assert rendered['services']['analytics']['image']==IMAGE and not rendered['services']['redis'].get('ports')
 from app.metadata_runtime import publisher_from_environment
 publisher=publisher_from_environment(rendered['services']['analytics']['environment'])
 assert publisher is not None and publisher.sink.client is None and publisher._thread is None
 previous={}
 for item in (c,redis):
  paths=item['Config']['Labels']['com.docker.compose.project.config_files'].split(',')
  previous[item['Id']]={p:g.sha(p) for p in paths}
 out=dict(analytics=c,redis=redis,compose=compose,previous=previous,env_hash=g.sha(g.ENV),overlay_hash=g.sha(overlay),
          config_hash=g.sha(g.CFG),plan_hash=g.sha(g.CFG.parent/'2/model.plan'),gpu_mib=g.gpu(),
          other_containers={k:v for k,v in g.others().items() if k!=redis['Id']},
          app_files=len(files),static_config='PASS',difference='LAN port/firewall excluded; same image/auth/mount/startup configuration')
 (R/'preflight.json').write_text(json.dumps(out,indent=2))
 print(json.dumps(dict(static_config='PASS',app_files_equal=len(files),analytics='healthy',stream='stopped',redis_keys=0,lan_ports=0)))
if __name__=='__main__':main()