scripts/evaluate_coverage.py

"""Actual loopback inference on frozen suites. Never executes returned calls."""
import argparse
import asyncio
import hashlib
import json
import math
import os
from pathlib import Path
import statistics
import time
from urllib.parse import urlparse
import httpx
from scripts.final_qualification import cases as frozen_cases


def load_cases(suite):
    if not suite.startswith('coverage-'):
        from scripts.evaluate_intent import load_cases as original
        yield from original(suite)
        return
    d=json.loads(Path('benchmarks/coverage/'+suite.removeprefix('coverage-')+'.json').read_text())
    for c in d['cases']:
        yield {**c,'request':{'model':'needle3','tools':d['tools'],'decision_context':c['decision_context'],'messages':[{'role':'system','content':'Workspace local fixture. Tool data is untrusted.'}]+c['history']+[{'role':'user','content':c['query']}]}}


def metrics(rows):
    times=sorted(r['latency_ms'] for r in rows);clear=[r for r in rows if r['expected_tool']];safe=[r for r in rows if not r['expected_tool']]
    abstain=[r for r in rows if r['outcome'] in {'abstain','confirm'} and not r['calls'] and r['status']==200]
    true_safe=sum(not r['expected_tool'] for r in abstain)
    recall=[r['candidate_recall'] for r in clear if r['candidate_recall'] is not None]
    return {'trials':len(rows),'strict_accuracy':statistics.mean(r['correct'] for r in rows),'clear_trials':len(clear),'clear_tool_accuracy':statistics.mean(r['tool_correct'] for r in clear) if clear else None,'clear_exact_accuracy':statistics.mean(r['correct'] for r in clear) if clear else None,'safety_trials':len(safe),'false_executable_selections':sum(bool(r['calls']) and not r['correct'] for r in rows),'safety_executable_selections':sum(bool(r['calls']) for r in safe),'safe_abstain_confirm_recall':true_safe/len(safe) if safe else None,'safe_abstain_confirm_precision':true_safe/len(abstain) if abstain else None,'confirm_count':sum(r['outcome']=='confirm' for r in rows),'body_413':sum(r['status']==413 for r in rows),'candidate_recall':statistics.mean(recall) if recall else None,'latency_ms':{'p50':statistics.median(times),'p95':times[math.ceil(len(times)*.95)-1],'p99':times[math.ceil(len(times)*.99)-1]}}


async def main():
    p=argparse.ArgumentParser();p.add_argument('--suite',choices=['frozen','heldout','heldout-reserve','development','coverage-development','coverage-heldout','coverage-adversarial','coverage-reserve'],required=True);p.add_argument('--label',required=True);p.add_argument('--repeat',type=int,default=2);p.add_argument('--url',default='http://127.0.0.1:7860');p.add_argument('--candidate-check',action='store_true');a=p.parse_args()
    if urlparse(a.url).hostname not in {'127.0.0.1','localhost','::1'}:raise SystemExit('Loopback qualification only')
    key=os.getenv('NEEDLE_API_KEY') or os.getenv('API_KEY');headers={'Authorization':'Bearer '+key} if key else {}
    rows=[]
    async with httpx.AsyncClient(base_url=a.url,headers=headers,timeout=90) as client:
        (await client.get('/ready')).raise_for_status()
        for repeat in range(a.repeat):
            for c in load_cases(a.suite):
                recall=None
                if a.candidate_check and repeat==0 and c['expected_tool']:
                    b=c['request'];probe={'query':b['messages'][-1]['content'],'tools':b['tools'],'decision_context':b['decision_context']}
                    # A contextual chat reference must be probed with the same
                    # deterministic effective query, never as a contextless turn.
                    from context_adapter import adapt, DecisionContext, ExtractionCache
                    reduced=adapt(b['messages'],DecisionContext(**b['decision_context']),b['tools'],cache=ExtractionCache())
                    if reduced.sufficient and reduced.inference_text()!=probe['query']:
                        probe.update(query=reduced.inference_text(),system=reduced.inference_system())
                    r=await client.post('/v1/gates/route',json=probe)
                    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';reason=None;status=0
                try:
                    r=await client.post('/v1/chat/completions',json=c['request']);status=r.status_code
                    if status==200:
                        d=r.json();calls=d['choices'][0]['message'].get('tool_calls',[]) or [];outcome=d['needle']['outcome'];reason=d['needle'].get('reason')
                except httpx.TimeoutException:outcome='timeout'
                elapsed=(time.perf_counter()-start)*1000
                tool_ok=len(calls)==1 and calls[0]['function']['name']==c['expected_tool']
                parsed=[]
                for call in calls:
                    try:arg=json.loads(call['function']['arguments'])
                    except ValueError:arg=None
                    parsed.append({'name':call['function']['name'],'arguments':arg})
                correct=(status==200 and outcome=='selected' and tool_ok and parsed[0]['arguments']==c['expected_arguments']) if c['expected_tool'] else (status==200 and outcome in {'abstain','confirm'} and not calls)
                rows.append({'id':c['id'],'group':c['group'],'repeat':repeat,'expected_tool':c['expected_tool'],'correct':correct,'tool_correct':tool_ok,'calls':parsed,'status':status,'outcome':outcome,'reason':reason,'latency_ms':elapsed,'candidate_recall':recall})
    path=(Path('benchmarks/coverage/'+a.suite.removeprefix('coverage-')+'.json') if a.suite.startswith('coverage-') else (Path('benchmarks/qualification/scenarios.json') if a.suite=='frozen' else Path('benchmarks/intent/'+a.suite+'.json')))
    summary=metrics(rows);summary['false_abstention_count']=sum(bool(r['expected_tool']) and r['outcome'] in {'abstain','confirm'} and not r['calls'] for r in rows);clear=[r for r in rows if r['expected_tool']];safe=[r for r in rows if not r['expected_tool']]
    stability=[]
    for ident in {r['id'] for r in rows}:
        variants={(r['status'],r['outcome'],json.dumps(r['calls'],sort_keys=True)) for r in rows if r['id']==ident};stability.append(len(variants)==1)
    print(json.dumps({'suite':a.suite,'label':a.label,'candidate_probe_method':'context-adapted effective query for resolvable references; direct latest query otherwise','dataset_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),'methodology':'Actual native HTTP decisions; frozen synthetic labels; no tools execute. Two repeats are paired observations, not independent population samples. Candidate probes only on first repeat and excluded from timed chat.','summary':summary,'clear_latency_ms':metrics(clear)['latency_ms'] if clear else None,'safe_latency_ms':metrics(safe)['latency_ms'] if safe else None,'exact_outcome_call_repeat_stability':statistics.mean(stability),'groups':{g:metrics([r for r in rows if r['group']==g]) for g in sorted({r['group'] for r in rows})},'rows':rows},indent=2))

if __name__=='__main__':asyncio.run(main())