Spaces:
Running
Running
File size: 3,270 Bytes
25a2c1c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | """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()
|