Spaces:
Running
Running
| """Reproduce public frozen cases locally; retain synthetic native/HTTP outputs. | |
| Historical qualification stored counters only. These are fresh reproductions, | |
| not recovered historical confidence scores. Never executes a tool or opens ports. | |
| """ | |
| import asyncio | |
| import json | |
| from pathlib import Path | |
| from fastapi.testclient import TestClient | |
| import gateway | |
| def main(): | |
| cases=json.loads(Path('benchmarks/qualification/scenarios.json').read_text())['cases'] | |
| chosen=[c for c in cases if c['id'] in {'ambiguous-0','ambiguous-2','ambiguous-5'}] | |
| original=gateway.scheduler.schedule | |
| native=[] | |
| async def observe(tools,system,text,priority): | |
| result=await original(tools,system,text,priority) | |
| native.append({'candidate_names':[t['name'] for t in tools], 'candidate_required_fields':{t['name']:t['parameters'].get('required',[]) for t in tools},'native_result':result}) | |
| return result | |
| gateway.scheduler.schedule=observe | |
| gateway.API_KEY='';gateway.AUTH_MODE='optional' | |
| rows=[] | |
| try: | |
| with TestClient(gateway.app) as client: | |
| for repeat in range(3): | |
| for case in chosen: | |
| before=len(native) | |
| r=client.post('/v1/chat/completions',json=case['request']) | |
| rows.append({'id':case['id'],'repeat':repeat,'user_text':case['request']['messages'][-1]['content'],'policy_arguments':case['request']['decision_context'].get('arguments',{}),'status':r.status_code,'http_response':r.json(),'inference':native[before:]}) | |
| finally:gateway.scheduler.schedule=original | |
| print(json.dumps({'methodology':'Fresh local actual native reproductions; synthetic public frozen inputs only. No historical raw native outputs existed. Host resources; not CPU benchmark.','rows':rows},indent=2)) | |
| if __name__=='__main__':main() | |