scripts/audit_native_output.py

"""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()