Spaces:
Running
Running
| """Owned authenticated CPU state/restart probes; no ports or selected tool execution.""" | |
| import hashlib | |
| import json | |
| import os | |
| from pathlib import Path | |
| import secrets | |
| import subprocess | |
| import time | |
| def main(): | |
| D=['docker','--host=unix:///var/run/docker.sock'];env=dict(os.environ,NEEDLE_API_KEY=secrets.token_hex(24)) | |
| for key in ['DOCKER_HOST','DOCKER_CONTEXT','DOCKER_TLS','DOCKER_TLS_VERIFY','DOCKER_CERT_PATH']:env.pop(key,None) | |
| def run(args):return subprocess.run(D+args,env=env,capture_output=True,text=True,check=True).stdout | |
| for label,image in [('before','needle-gateway-v2:intent-guard'),('after','needle-gateway-v2:coverage')]: | |
| cid=run(['run','-d','--cpus','2','--memory','16g','--network','none','-e','NEEDLE_API_KEY','-e','NEEDLE_AUTH_MODE=required','--entrypoint','uvicorn',image,'gateway:app','--app-dir','/srv','--host','127.0.0.1','--port','7860']).strip() | |
| try: | |
| checked={} | |
| for file in ['gateway.py','models.py','registry.py','context_adapter.py','decision_policy.py','intent_guard.py','request_limits.py','laya.py']: | |
| source=subprocess.check_output(['git','show','9c3db09:'+file]) if label=='before' else Path(file).read_bytes() | |
| actual=run(['exec',cid,'sha256sum','/srv/'+file]).split()[0] | |
| assert actual==hashlib.sha256(source).hexdigest() | |
| checked[file]=actual | |
| model=run(['exec','--workdir','/root/.cache/cactus-needle/v3/3.0.1',cid,'sha256sum','--check','/srv/needle-assets.sha256']) | |
| for folder in ['scripts','benchmarks']:run(['cp',folder,cid+':/srv/'+folder]) | |
| def ready(): | |
| for _ in range(50): | |
| try: | |
| run(['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']}))"]);return | |
| except subprocess.CalledProcessError:time.sleep(.2) | |
| raise RuntimeError('Owned test server not ready') | |
| ready();smoke=json.loads(run(['exec',cid,'python','-m','scripts.smoke_registry'])) | |
| initialize=json.loads(run(['exec',cid,'python','-m','scripts.qualify_restart','initialize'])) | |
| start=time.perf_counter();run(['restart',cid]);ready();restart_ms=(time.perf_counter()-start)*1000 | |
| recovered=json.loads(run(['exec',cid,'python','-m','scripts.qualify_restart','after-restart'])) | |
| output={'methodology':'Actual authenticated local CPU registry/state/restart smoke, synthetic repeated requests; no returned tool executes. Not HF or held-out accuracy.','resources':{'cpu_quota':2,'memory_gib':16,'network':'none','published_ports':[],'client_in_same_cgroup':True},'verified_runtime_source_sha256':checked,'model_asset_check':model.strip().splitlines(),'state_smoke':smoke,'restart_initialize':initialize,'restart_recovery':recovered,'restart_to_authorized_ready_ms':restart_ms} | |
| Path('benchmarks/coverage/results/'+label+'-state-cpu.json').write_text(json.dumps(output,indent=2)+'\n') | |
| print(label,smoke['warm_delta_latency_ms'],restart_ms,flush=True) | |
| finally:run(['rm','-f',cid]) | |
| if __name__=='__main__':main() | |