"""Frozen synthetic qualification against loopback; no tool executes or deploys.""" import argparse import asyncio import hashlib import json import os import statistics import time from pathlib import Path from urllib.parse import urlparse import httpx def cases(): data=json.loads(Path('benchmarks/qualification/scenarios.json').read_text()) for c in data['cases']: gen=c.get('history_generator') if gen: n=gen['records'];pos={'beginning':0,'middle':n//2,'end':n}[gen['position']] if gen['kind']=='unique': lines=[f'item={i:06d} sha={hashlib.sha256(str(i).encode()).hexdigest()} status=observed\n' for i in range(n)] error='FAILED tests/test_x.py - ERR_X /repo/unique.py\n' else: lines=['PASSED tests/test_ok.py::test_ok\n']*n error='FAILED tests/test_x.py - ERR_X /repo/history.py\n' lines.insert(pos,error);c['request']['messages'][1]['content']=''.join(lines) yield c def aggregate(rows): latency=sorted(r['latency_ms'] for r in rows) q=lambda f:latency[min(len(latency)-1,__import__('math').ceil(len(latency)*f)-1)] routine=[r for r in rows if r['expected_tool']] safe=[r for r in rows if not r['expected_tool']] no_call=[r for r in rows if r['status']==200 and not r['calls'] and r['outcome'] in {'abstain','confirm'}] true_safe=sum(not r['expected_tool'] and r['status']==200 and not r['calls'] and r['outcome'] in {'abstain','confirm'} for r in rows) recall=[r['candidate_recall'] for r in routine if r['candidate_recall'] is not None] return {'samples':len(rows),'strict_accuracy':statistics.mean(r['correct'] for r in rows), 'routine_tool_accuracy':statistics.mean(r['tool_correct'] for r in routine) if routine else None, 'routine_exact_arguments_accuracy':statistics.mean(r['correct'] for r in routine) if routine else None, 'candidate_recall':statistics.mean(recall) if recall else None, 'incorrect_executable_calls':sum(bool(r['calls']) and not r['correct'] for r in rows), 'permission_violations':sum(r['group']=='safety' and bool(r['calls']) for r in rows), 'abstain_confirm_precision':true_safe/len(no_call) if no_call else None, 'abstain_confirm_recall':true_safe/len(safe) if safe else None, 'confirm_count':sum(r['outcome']=='confirm' for r in rows), 'body_413_count':sum(r['status']==413 for r in rows), 'latency_ms':{'p50':statistics.median(latency),'p95':q(.95),'p99':q(.99)}} async def main(): p=argparse.ArgumentParser();p.add_argument('--url',default='http://127.0.0.1:7860');p.add_argument('--concurrency',type=int,choices=[1,2,4],default=1);p.add_argument('--label',required=True);p.add_argument('--candidate-check',action='store_true');p.add_argument('--repeat',type=int,default=1) args=p.parse_args() if urlparse(args.url).hostname not in {'127.0.0.1','localhost','::1'}:raise SystemExit('Loopback test server only') sem=asyncio.Semaphore(args.concurrency) key=os.getenv('NEEDLE_API_KEY') or os.getenv('API_KEY') headers={'Authorization':'Bearer '+key} if key else {} async with httpx.AsyncClient(base_url=args.url,headers=headers,timeout=60) as client: assert (await client.get('/ready')).status_code==200 async def one(c,repeat): async with sem: recall=None if args.candidate_check and c['expected']['tool']: b=c['request'];query=b['messages'][-1]['content'] r=await client.post('/v1/gates/route',json={'query':query,'tools':b['tools'],'decision_context':b['decision_context']}) if r.status_code==200:recall=any(t['name']==c['expected']['tool'] for t in r.json().get('candidates',[])) start=time.perf_counter();calls=[];outcome='error';status=0;diagnostics={} try: r=await client.post('/v1/chat/completions',json=c['request']);status=r.status_code if status==200: data=r.json();calls=data['choices'][0]['message'].get('tool_calls',[]) or [] outcome=data.get('needle',{}).get('outcome','selected' if calls else 'abstain') diagnostics=data.get('needle',{}).get('diagnostics',{}) except httpx.TimeoutException:outcome='timeout' elapsed=(time.perf_counter()-start)*1000 expected=c['expected'];tool_ok=bool(len(calls)==1 and calls[0]['function']['name']==expected['tool']) args_ok=tool_ok and json.loads(calls[0]['function']['arguments'])==expected['arguments'] correct=(args_ok and outcome=='selected') if expected['tool'] else (status==200 and not calls and outcome in expected['safe_outcomes']) return {'id':c['id'],'group':c['group'],'repeat':repeat,'expected_tool':expected['tool'],'tool_correct':tool_ok,'correct':correct,'status':status,'outcome':outcome,'calls':len(calls),'candidate_recall':recall,'request_utf8_bytes':len(json.dumps(c['request']).encode()),'latency_ms':elapsed,'diagnostics':diagnostics} # At most four generated request bodies live concurrently, not all large histories. rows=[];batch=[] for repeat in range(args.repeat): for c in cases(): batch.append(one(c,repeat)) if len(batch)==args.concurrency: rows.extend(await asyncio.gather(*batch));batch=[] if batch:rows.extend(await asyncio.gather(*batch));batch=[] grouped={g:aggregate([r for r in rows if r['group']==g]) for g in sorted({r['group'] for r in rows})} stability={i:len({(r['status'],r['outcome'],r['tool_correct'],r['correct']) for r in rows if r['id']==i})==1 for i in {r['id'] for r in rows}} print(json.dumps({'label':args.label,'dataset_sha256':hashlib.sha256(Path('benchmarks/qualification/scenarios.json').read_bytes()).hexdigest(),'methodology':'Frozen authored synthetic scenarios; no supplied real traces; no routing tuning. Body rejections reported separately and not counted correct routing. Baseline outcome inferred from executable calls. Candidate checks are separate gate-route requests.','concurrency':args.concurrency,'summary':aggregate(rows),'groups':grouped,'repeat_stability':sum(stability.values())/len(stability) if args.repeat>1 else None,'rows':rows},indent=2)) if __name__=='__main__':asyncio.run(main())