intelligence-radar/intelligence_radar/trials.py

"""Budgeted OpenRouter work-sample trials with deterministic verification."""
from datetime import datetime, timezone
from decimal import Decimal
import fcntl
import hashlib
import json
import os
from pathlib import Path
import time
from urllib.request import Request, urlopen

from .store import all_records, canonical, get, put

ENDPOINT = 'https://openrouter.ai/api/v1/chat/completions'


def money(value):
    result = Decimal(str(value))
    if not result.is_finite() or result < 0:
        raise ValueError('invalid nonnegative dollar amount')
    return result


def budget(root, campaign):
    events = [(i,e) for i,e in all_records(root,'trial-events') if e['campaign']==campaign]
    reservations = {i:e for i,e in events if e['kind']=='reserved'}
    settled = {e['reservation_id']:e for _,e in events if e['kind']=='settled'}
    spent = sum((money(e['actual_usd']) for e in settled.values()),Decimal(0))
    pending = sum((money(e['reserved_usd']) for i,e in reservations.items() if i not in settled),Decimal(0))
    return {'spent_usd':str(spent), 'pending_usd':str(pending), 'committed_usd':str(spent+pending)}


def verify(case, content):
    try:
        actual=json.loads(content)
    except (ValueError,TypeError):
        return {'accepted':False,'reason':'response is not JSON','actual':None}
    # Canonical JSON comparison distinguishes booleans from numeric values.
    accepted=canonical(actual)==canonical(case['expected'])
    return {'accepted':accepted,'reason':'exact structured match' if accepted else 'structured mismatch', 'actual':actual}


def plan_trial(case, model):
    prompt=case['instruction']+'\nInput:\n'+json.dumps(case['input'],sort_keys=True)
    rates=model['max_usd_per_million']
    payload={'model':model['id'],'messages':[{'role':'user','content':prompt}],
             'max_tokens':2048,'stream':False,'usage':{'include':True},
             'provider':{'allow_fallbacks':False,'require_parameters':True,
                         'max_price':{'prompt':float(money(rates['prompt'])),
                                      'completion':float(money(rates['completion'])),'request':0}}}
    # UTF-8 bytes plus a conservative chat framing allowance bound these short text prompts.
    reserved=(Decimal(len(prompt.encode('utf-8'))+1024)*money(rates['prompt'])+
              Decimal(payload['max_tokens'])*money(rates['completion']))/Decimal(1000000)
    return payload,reserved


def trial_id(campaign, case, payload):
    return hashlib.sha256(canonical({'campaign':campaign,'case':case,'payload':payload})).hexdigest()


def plan(root, config, suite, pilot=False):
    """Render review inputs without credentials, network calls or ledger writes."""
    campaign = config['campaign']
    limit = money(config['budget_usd'])
    for label, items in [('case', suite['cases']), ('model', config['models'])]:
        ids = [item['id'] for item in items]
        if not ids or len(ids) != len(set(ids)):
            raise ValueError(f'{label} IDs must be nonempty and unique')
    events = [(i, e) for i, e in all_records(root, 'trial-events') if e['campaign'] == campaign]
    held = {i: e for i, e in events if e['kind'] == 'reserved'}
    settled = {e['reservation_id']: e for _, e in events if e['kind'] == 'settled'}
    spent = sum((money(e['actual_usd']) for e in settled.values()), Decimal(0))
    pending = sum((money(e['reserved_usd']) for i, e in held.items() if i not in settled), Decimal(0))
    existing = {e['run_id']: 'settled' if i in settled else 'reserved' for i, e in held.items()}
    rows = []
    new_total = Decimal(0)
    for case in suite['cases'][:2] if pilot else suite['cases']:
        for model in config['models']:
            payload, reserved = plan_trial(case, model)
            ident = trial_id(campaign, case, payload)
            status = existing.get(ident, 'new')
            if status == 'new':
                new_total += reserved
            rows.append({'case_id': case['id'], 'model': model['id'], 'run_id': ident,
                         'status': status, 'reserved_usd': str(reserved),
                         'request_sha256': hashlib.sha256(canonical(payload)).hexdigest()})
    blockers = []
    if pending:
        blockers.append('unresolved billing reservations')
    if any(money(e['actual_usd']) > money(held[i]['reserved_usd']) for i, e in settled.items()):
        blockers.append('campaign halted after provider cost exceeded reservation')
    total = spent + pending + new_total
    if total > limit:
        blockers.append('planned reservations exceed campaign budget')
    return {'schema_version': 1, 'campaign': campaign, 'pilot': pilot,
            'execution_authorized': False,
            'scope': 'review inputs only; requires admitted recipient, credential delivery and billing reconciliation',
            'config_sha256': hashlib.sha256(canonical(config)).hexdigest(),
            'suite_sha256': hashlib.sha256(canonical(suite)).hexdigest(),
            'budget_usd': str(limit), 'spent_usd': str(spent), 'pending_usd': str(pending),
            'new_reservations_usd': str(new_total), 'planned_committed_usd': str(total),
            'budget_blockers': blockers, 'trials': rows}


def execute(root, config, case, model, opener=urlopen):
    root=Path(root)
    if not os.environ.get('OPENROUTER_API_KEY'):
        raise ValueError('OPENROUTER_API_KEY is required')
    campaign=config['campaign']
    limit=money(config['budget_usd'])
    payload,reserved=plan_trial(case,model)
    run_id=trial_id(campaign, case, payload)
    lockpath=root/'.runtime/trials.lock';lockpath.parent.mkdir(parents=True,exist_ok=True)
    with lockpath.open('a') as lock:
        fcntl.flock(lock,fcntl.LOCK_EX)
        events = all_records(root, 'trial-events')
        reservations = {i: e for i, e in events if e['kind'] == 'reserved'}
        for _, event in events:
            if event['campaign'] == campaign and event['kind'] == 'settled':
                held = reservations[event['reservation_id']]
                if money(event['actual_usd']) > money(held['reserved_usd']):
                    raise ValueError('campaign halted: provider cost exceeded reservation')
        for ident,event in all_records(root,'trial-events'):
            if event['campaign']==campaign and event.get('run_id')==run_id:
                if event['kind']=='settled':
                    return get(root,'trials',event['trial_id'])
        if any(e.get('run_id')==run_id for _,e in all_records(root,'trial-events')):
            raise ValueError('trial already reserved; reconcile uncertain billing before retrying')
        state=budget(root,campaign)
        if money(state['pending_usd']):
            raise ValueError('campaign has unresolved reservations; reconcile billing before continuing')
        if money(state['committed_usd'])+reserved>limit:
            raise ValueError('trial would exceed campaign budget')
        stamp=datetime.now(timezone.utc).isoformat()
        reservation=put(root,'trial-events',{'kind':'reserved','campaign':campaign,'run_id':run_id,
                                             'reserved_usd':str(reserved),'budget_usd':str(limit),
                                             'created_at':stamp,'model':model['id'],'case_id':case['id']})
        request=Request(ENDPOINT,data=json.dumps(payload).encode(),method='POST',
                        headers={'Content-Type':'application/json','Authorization':'Bearer '+os.environ['OPENROUTER_API_KEY'],
                                 'X-OpenRouter-Title':'Intelligence Radar field pilot'})
        start=time.monotonic()
        # No automatic inference retries: a timeout may already have incurred a charge.
        with opener(request,timeout=90) as response:
            raw=response.read(2*1024*1024+1)
        if len(raw)>2*1024*1024:
            raise ValueError('provider response exceeds retention limit; reservation retained')
        value=json.loads(raw)
        if 'error' in value:
            raise ValueError('provider returned an error; reservation retained')
        if value.get('usage',{}).get('cost') is None:
            raise ValueError('provider omitted billing cost; reservation retained')
        cost=money(value['usage']['cost'])
        content=value['choices'][0]['message'].get('content')
        trial={'schema_version':1,'campaign':campaign,'run_id':run_id,'case':case,
               'requested_model':model['id'],'returned_model':value.get('model'),
               'provider':value.get('provider'),'request':payload,'response':value,
               'verification':verify(case,content),'actual_usd':str(cost),
               'runtime_seconds':time.monotonic()-start,'completed_at':datetime.now(timezone.utc).isoformat(),
               'reservation_id':reservation,'synthetic':False,
               'sample_kind':'controlled repository work sample; real provider execution'}
        ident=put(root,'trials',trial)
        put(root,'trial-events',{'kind':'settled','campaign':campaign,'run_id':run_id,
                                 'reservation_id':reservation,'trial_id':ident,'actual_usd':str(cost)})
        if cost>reserved:
            raise ValueError('provider cost exceeded reservation; campaign halted for review')
        return trial


def run(root, config, suite, pilot=False):
    rows=[]
    selected=suite['cases'][:2] if pilot else suite['cases']
    for case in selected:
        for model in config['models']:
            trial=execute(root,config,case,model)
            rows.append({'case':case['id'],'model':model['id'],
                         'accepted':trial['verification']['accepted'],'actual_usd':trial['actual_usd']})
            print(json.dumps(rows[-1]),flush=True)
    return {'campaign':config['campaign'],'trials':rows,'budget':budget(root,config['campaign'])}