"""Reproducible before/after local CPU qualification. Owns every container. Never publishes ports, deploys remotely, or executes selected tools. Expected before runtime files are verified against the exact requested Git commit. """ import argparse import hashlib import json import os from pathlib import Path import subprocess import time def main(): p=argparse.ArgumentParser();p.add_argument('--labels',nargs='+',choices=['before','after'],default=['before','after']);p.add_argument('--suites',nargs='+',default=['frozen','heldout','heldout-reserve','coverage-development','coverage-heldout','coverage-adversarial','coverage-reserve','workflow-development','workflow-heldout','workflow-schema-development','workflow-schema-heldout']);p.add_argument('--no-tests',action='store_true');p.add_argument('--reuse-baseline',action='store_true');p.add_argument('--before-image',default='needle-gateway-v2:coverage');p.add_argument('--after-image',default='needle-gateway-v2:workflow');a=p.parse_args() import secrets env=dict(os.environ,NEEDLE_API_KEY=secrets.token_hex(24)) for k in ['DOCKER_HOST','DOCKER_CONTEXT','DOCKER_TLS','DOCKER_TLS_VERIFY','DOCKER_CERT_PATH']:env.pop(k,None) D=['docker','--host=unix:///var/run/docker.sock'] def run(args,**kw):return subprocess.run(D+args,env=env,capture_output=True,text=True,check=True,**kw) out=Path('benchmarks/workflow/results');out.mkdir(exist_ok=True) runtime=['gateway.py','models.py','registry.py','laya.py','context_adapter.py','decision_policy.py','request_limits.py'] for label,image,commit in [('before',a.before_image,'ef25f06c0cebea319abaed86f49dffacd96938ce'),('after',a.after_image,None)]: if label not in a.labels:continue start=time.perf_counter() cid=run(['create','--cpus','2','--memory','16g','--network','none','-e','NEEDLE_API_KEY','-e','NEEDLE_AUTH_MODE=required','--entrypoint','uvicorn',image,'scripts.workflow_cpu_app:app','--app-dir','/srv','--host','127.0.0.1','--port','7860']).stdout.strip() try: for folder in ['scripts','benchmarks']:run(['cp',folder,cid+':/srv/'+folder]) run(['start',cid]) checked={} for file in runtime+['intent_guard.py']+(['payload_transport.py','native_schema.py'] if label=='after' else []): expected=subprocess.check_output(['git','show',commit+':'+file]) if commit else Path(file).read_bytes() actual=run(['exec',cid,'sha256sum','/srv/'+file]).stdout.split()[0] assert hashlib.sha256(expected).hexdigest()==actual,(label,file,'source mismatch') checked[file]=actual model_check=run(['exec','--workdir','/root/.cache/cactus-needle/v3/3.0.1',cid,'sha256sum','--check','/srv/needle-assets.sha256']).stdout ready=False for _ in range(50): r=subprocess.run(D+['exec',cid,'python','-c',"import os,urllib.request;urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:7860/ready',headers={'Authorization':'Bearer '+os.environ['NEEDLE_API_KEY']}))"],env=env,capture_output=True) if not r.returncode:ready=True;break time.sleep(.2) if not ready:raise RuntimeError('Owned test container did not become ready') startup_ms=(time.perf_counter()-start)*1000 for suite in a.suites: if label=='before' and a.reuse_baseline and (out/(label+'-'+suite+'.json')).exists(): cached=json.loads((out/(label+'-'+suite+'.json')).read_text()) if cached.get('resources',{}).get('sdk_date_fact')!='date: 2026-10-17 Sat 00:00':raise RuntimeError('Refusing baseline reuse across different SDK clock conditions') assert cached['verified_runtime_source_sha256']==checked fixture=Path('benchmarks/workflow/'+suite.removeprefix('workflow-')+'.json') if suite.startswith('workflow-') else Path('benchmarks/coverage/'+suite.removeprefix('coverage-')+'.json') if suite.startswith('coverage-') else (Path('benchmarks/qualification/scenarios.json') if suite=='frozen' else Path('benchmarks/intent/'+suite+'.json')) assert cached['dataset_sha256']==hashlib.sha256(fixture.read_bytes()).hexdigest() continue if suite in {'heldout-reserve','coverage-development','coverage-heldout','coverage-adversarial','coverage-reserve','workflow-development','workflow-heldout','workflow-stress','workflow-schema-development','workflow-schema-heldout'}: run(['restart',cid]) for _ in range(50): probe=subprocess.run(D+['exec',cid,'python','-c',"import os,urllib.request;urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:7860/ready',headers={'Authorization':'Bearer '+os.environ['NEEDLE_API_KEY']}))"],env=env,capture_output=True) if not probe.returncode:break time.sleep(.2) else:raise RuntimeError('Reserve-suite restart failed') data=json.loads(run(['exec',cid,'python','-m','scripts.evaluate_workflow','--suite',suite,'--label',label,'--repeat','2','--candidate-check'],timeout=1200).stdout) data['server_memory']=json.loads(run(['exec',cid,'python','-c',"import json;print(json.dumps({l.split(':')[0]:l.split(':')[1].strip() for l in open('/proc/1/status') if l.startswith(('VmHWM:','VmRSS:','Threads:'))}))"]).stdout) data['verified_runtime_source_sha256']=checked data['image_id']=run(['image','inspect','--format','{{.Id}}',image]).stdout.strip() data['model_asset_check']=model_check.strip().splitlines() data['container_start_to_ready_ms']=startup_ms data['resources']={'cpu_quota':2,'memory_gib':16,'network':'none','published_ports':[],'client_in_same_cgroup':True,'sdk_date_fact':'date: 2026-10-17 Sat 00:00','clock_control_sha256':hashlib.sha256(Path('scripts/workflow_cpu_app.py').read_bytes()).hexdigest()} (out/(label+'-'+suite+'.json')).write_text(json.dumps(data,indent=2)+'\n') print(label,suite,json.dumps(data['summary']),flush=True) finally:run(['rm','-f',cid]) if a.no_tests:return tests=run(['run','--rm','--cpus','2','--memory','16g','--network','none','needle-gateway-v2:workflow-tests']) (out/'offline-cpu-tests.txt').write_text(tests.stdout+tests.stderr) print(tests.stdout,flush=True) if __name__=='__main__':main()