| """Real callback-driven escalation. Internal budget/escalation layer: the answer |
| contract above it always answers and refuses to emit any deferral field.""" |
| from dataclasses import dataclass |
| import math |
| from numbers import Integral, Real |
|
|
| @dataclass(frozen=True) |
| class Budget: |
| branches: int = 8 |
| input_tokens: int = 8192 |
| generated_tokens: int = 3072 |
|
|
|
|
| def route(stages, budget=Budget()): |
| """stages are (name, reserved_cost, callback). Callback returns answer/confidence/cost. |
| |
| Reservations are hard maxima enforced by the backend, not optimistic estimates. |
| Callbacks must honor the reservation passed to them. Overspending fails closed. |
| """ |
| used = {'branches':0,'input_tokens':0,'generated_tokens':0} |
| valid_cost=lambda value:isinstance(value,Integral) and not isinstance(value,bool) and value>=0 |
| valid_probability=lambda value:isinstance(value,Real) and not isinstance(value,bool) and math.isfinite(value) and 0<=value<=1 |
| if any(not valid_cost(getattr(budget,k)) for k in used): |
| raise ValueError('invalid compute budget') |
| traces, last = [], None |
| reasons = {'fast':'full_depth','views':'escalated_views','reasoning':'escalated_reasoning','exit':'early_exit'} |
| for name,reserve,call in stages: |
| if name not in reasons or set(reserve)!=set(used) or any(not valid_cost(v) for v in reserve.values()): |
| raise ValueError('invalid stage reservation') |
| if any(used[k]+reserve.get(k,0) > getattr(budget,k) for k in used): |
| return {'answer':None,'candidate':last,'abstained':True,'compute':{'stop_reason':'budget_exhausted','cost':used,'stages':traces}} |
| result = call(dict(reserve)) |
| cost = result['cost'] |
| if set(cost)!=set(used) or any(not valid_cost(cost[k]) or cost[k] > reserve[k] for k in used): |
| raise ValueError('backend exceeded reserved compute budget') |
| for k in used: |
| used[k] += cost.get(k,0) |
| last = result.get('answer') |
| confidence = result.get('confidence') or {} |
| if not isinstance(confidence,dict): |
| confidence={} |
| traces.append({'stage':name,'cost':cost,'confidence':confidence}) |
| certified = confidence.get('calibration_support') == 'in_validated_scope' and confidence.get('policy_status') == 'confirmed' |
| accepted = (certified and last is not None and valid_probability(confidence.get('reliability')) |
| and valid_probability(confidence.get('threshold')) and confidence['reliability'] >= confidence['threshold']) |
| if accepted: |
| return {'answer':last,'abstained':False,'confidence':confidence,'compute':{'stop_reason':reasons[name],'cost':used,'stages':traces}} |
| return {'answer':None,'candidate':last,'abstained':True,'compute':{'stop_reason':'abstain','cost':used,'stages':traces}} |
|
|
|
|
| def forced_answer_suffix(trace): |
| """Backend appends this to a capped trace and performs a letter-logit forward.""" |
| return ('' if '</think>' in trace else '\n</think>') + '\nAnswer (one letter):' |
|
|
