Spaces:
Running
Running
| """Public synthetic native-output development probes, never tool execution.""" | |
| import copy,json,time,resource | |
| from pathlib import Path | |
| import needle | |
| from context_adapter import DecisionContext,adapt,ExtractionCache,budget_fits | |
| from gateway import with_current_facts | |
| from decision_policy import validate_decision | |
| def main(): | |
| original=json.loads(Path('benchmarks/coverage/results/after-workflow.json').read_text())['rows'][5]['request'] | |
| cases=[('small','hello'),('original_multiline','def add(a, b):\n return a + b\n'),('medium','x'*256),('unicode','Grüße 世界 🧪\n'*8),('escaping','path = "C:\\tmp\\x"\n'*8),('json_string',json.dumps({'nested':{'items':['a','b']}})),('nested_object',{'level':{'value':'data'}})] | |
| rows=[] | |
| for name,content in cases: | |
| tools=copy.deepcopy(original['tools']);tool=next(t for t in tools if t['name']=='write_file') | |
| if isinstance(content,dict):tool['parameters']['properties']['content']={'type':'object','properties':{'level':{'type':'object'}},'required':['level']} | |
| args={'path':'/project/app.py','content':content};policy=DecisionContext(workspace='/project',arguments={'write_file':args}) | |
| query='write_file /project/app.py content '+json.dumps(content,ensure_ascii=False) | |
| system=with_current_facts('Public disposable repository.',[tool],policy) | |
| agent=needle.Needle(tools=[tool],system=system,generation=3) | |
| start=time.perf_counter();native=agent.complete(query);elapsed=(time.perf_counter()-start)*1000 | |
| valid=validate_decision(native,[tool],policy,system+'\n'+query) | |
| rows.append({'id':name,'content_utf8_bytes':len(json.dumps(content,ensure_ascii=False).encode()),'schema_utf8_bytes':len(json.dumps(tool['parameters']).encode()),'prompt_proxy_units':budget_fits(query,system,[tool],10**9)[1],'public_max_new_tokens':512,'latency_ms':elapsed,'native':native,'validated':valid,'serialized_envelope_bytes':len(json.dumps(native).encode()),'host_self_peak_rss_kib':resource.getrusage(resource.RUSAGE_SELF).ru_maxrss}) | |
| # A separate fresh native decision on a bounded private path-only schema. | |
| tool=copy.deepcopy(next(t for t in original['tools'] if t['name']=='write_file')) | |
| tool['parameters']['properties'].pop('content');tool['parameters']['required'].remove('content') | |
| policy=DecisionContext(workspace='/project',arguments={'write_file':{'path':'/project/app.py'}}) | |
| query='write_file /project/app.py';system=with_current_facts('Explicit current action; payload is request-local inert data.',[tool],policy) | |
| agent=needle.Needle(tools=[tool],system=system,generation=3);start=time.perf_counter();native=agent.complete(query) | |
| projection={'tools':[tool],'query':query,'latency_ms':(time.perf_counter()-start)*1000,'native':native,'validated':validate_decision(native,[tool],policy,system+'\n'+query)} | |
| Path('benchmarks/workflow/results/native-audit.json').write_text(json.dumps({'methodology':'Fresh host native probes on development cases; unchanged official 3.0.1 model and supported SDK API only. No executable tool runs. Proxy units are not exact native tokens; serialized envelope length is not hidden generated token length.','rows':rows,'private_projection_probe':projection},indent=2)+'\n') | |
| print([(r['id'],r['native'].get('success'),r['native'].get('confidence')) for r in rows]);print('projection',projection['native']) | |
| if __name__=='__main__':main() | |