alee/6krrt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
neuralwatt-router-service
6krrt/dispatcher.py
adlee-was-taken 943b2d5cde fix: resolve TUI live-routing panel review findings
2026-08-24 00:22:01 -04:00

2635 lines
104 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""FastAPI dispatcher for the local LLM model router.
Pipeline per request:
task text
-> local classifier (Ollama) category / tier / context estimate
-> escalation low confidence bumps the tier
-> hard filters (routing.py) context, tier, freshness, access, latency
-> weighted score (scoring.py) cost / eco / proficiency
-> dispatch to NeuralWatt OpenAI-compatible chat completion
-> energy_observations real cost + energy for the call
Run:
uvicorn dispatcher:app --reload
Endpoints:
GET /health provider/classifier reachability, catalog counts
POST /route classify and pick a model, but do NOT call it
POST /dispatch /route, then call the winning model and log it
GET /v1/models OpenAI-compatible catalog
POST /v1/chat/completions OpenAI-compatible, routes then proxies
/route exists because every part of this is testable without spending a
token, and a routing bug is much easier to see in a ranked candidate list
than in a completion.
The /v1 pair is what a normal OpenAI client (opencode, an SDK, curl) talks
to. Ask for the model `auto` and the router chooses; ask for `auto:batch`
and it also admits flex rows; ask for a real model id and it goes straight
there, still logged.
"""
from __future__ import annotations
import json
import os
import hashlib
import queue
import re
import sqlite3
import time
from datetime import datetime, timezone
from statistics import median
from typing import Any, Literal, Optional
import requests
import secrets
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, HTTPException
from fastapi.responses import JSONResponse, Response, StreamingResponse
from openai import OpenAI, OpenAIError
from pydantic import BaseModel, Field
import logs
from capabilities import detect_capabilities, iter_image_url_values
from config import RouterConfig, load_config
from context_prune import extract_text, prune_context
import events
from routing import (
BATCH,
INTERACTIVE,
capability_gate_reason,
rank_candidates,
rejection_reason,
select_candidates,
)
from iteration import attempts_allowed, plan_retry
from verification import (
LOCAL_VERIFY_SYSTEM,
build_local_verify_prompt,
interpret_local_verdict,
parse_verdict_json,
verify_response,
worth_local_check,
)
from metrics import (
per_model,
quota_burn,
recent_decisions,
scoring_coverage,
top_proficiency,
verdict_mix,
)
# Virtual model names that mean "you pick". Anything else is taken as a real
# model id and dispatched as asked.
ROUTER_MODEL = "auto"
ROUTER_MODEL_BATCH = "auto:batch"
# Observations from the fixed reference workload in seed_energy.py. Only
# these steer routing; organic traffic is logged for accounting but varies
# too much with request shape to compare models by.
SEED_CATEGORY = "seed_reference"
# NeuralWatt reports this when it cannot resolve live grid data and falls back
# to a constant (475.0 gCO2/kWh) while still echoing the original grid_id. The
# figure is a placeholder, not a measurement, so it must not steer eco scoring.
FALLBACK_CARBON_SOURCE = "static_fallback"
# Measured: cost_usd / energy_kwh came back 8.00 across every model. Kept for
# reporting and for reasoning about bills; scoring reads billed cost directly
# and so does not need it.
USD_PER_KWH = 8.00
def gross_energy_kwh(avg_power_watts: float, duration_seconds: float) -> float:
"""Pool energy over a request, BEFORE multi-tenant attribution.
Diagnostic only — scoring does NOT use this. It exists to make the
decomposition inspectable: billed ``energy_kwh`` equals this times
``attribution_ratio``, so comparing the two shows how much of a model's
cost comes from serving concurrency rather than from the work itself.
Ranking on this figure was tried and is wrong: it strips a term that
varies 750x between models (real, and real money) to suppress one that
varies about 1.8x within a model (noise).
"""
return avg_power_watts * duration_seconds / 3_600_000
# Rough chars-per-token, used as a FLOOR on the classifier's context estimate:
# a real coding session sends the whole conversation, and the classifier —
# which sees that conversation and is asked to estimate its own input —
# underestimates it by orders of magnitude (10 tokens for a 600-token exchange
# in testing). So the measured size wins when it is larger.
#
# Deliberately conservative at 3, not the ~4 that holds for English prose.
# Measured from real opencode traffic: a 106,158-token prompt was routed to a
# model with a 94,196 effective window, meaning the true density was at most
# 3.55 chars/token — agent traffic is code, JSON and tool schemas, which pack
# far denser than prose. That overshot the guard by ~12k tokens and survived
# only on the 0.75 safety factor's slack.
#
# The asymmetry matters: underestimating silently admits a model that cannot
# hold the prompt, while overestimating merely picks a roomier one.
CHARS_PER_TOKEN = 3
load_dotenv()
# kWh -> BTU. Serves no routing purpose; the design doc wants it on the
# dashboard, so it is stored alongside the kWh figure rather than derived.
BTU_PER_KWH = 3412.14
app = FastAPI(title="Local LLM Model Router", version="0.1.0")
cfg: RouterConfig = load_config("config.yaml")
logs.configure(cfg.logging.level)
# --- request/response models ---------------------------------------------
class TaskRequest(BaseModel):
task: str = Field(..., description="The task to route.")
context: Optional[str] = Field(
None,
description=(
"Assembled context (docs/code) to send with the task. "
"On dispatch_endpoint the chat path reuses this field to carry "
"the prior conversation turn (framing) passed to the classifier; "
"this dual use is safe because dispatch_endpoint only reads it "
"for the classify() branch, while the chat path rewrites the "
"Classification when task_category+task_tier+required_context "
"are all provided (the override branch never reads req.context)."
),
)
latency_tolerance: Optional[Literal["interactive", "batch"]] = Field(
None,
description=(
"'batch' admits flex rows, which are held server-side during peak. "
"Defaults to routing.default_latency_tolerance."
),
)
tools_present: bool = Field(
False,
description=(
"Whether the caller has tool definitions on the table. Requires a "
"model with measured tool-use competence (routing."
"min_tool_proficiency). Set automatically from the `tools` array "
"on /v1/chat/completions; exposed here for testing and for "
"clients that route before they call."
),
)
has_images: bool = Field(
False,
description="Whether the request carries image parts; requires a "
"vision-capable model (routing.require_vision). Set automatically from "
"the request body on /v1/chat/completions.",
)
require_json_mode: bool = Field(
False,
description="Whether the request's response_format needs JSON mode; "
"requires a JSON-mode-capable model (routing.require_json_mode).",
)
# Overrides, mostly for testing the router without the classifier in the loop.
task_category: Optional[str] = None
task_tier: Optional[int] = Field(None, ge=1, le=3)
required_context_tokens: Optional[int] = Field(None, ge=0)
class Classification(BaseModel):
task_category: str
task_tier: int
required_context_tokens: int
confidence: float
escalated: bool = False
source: Literal["classifier", "override", "fallback"] = "classifier"
class Candidate(BaseModel):
model_id: str
provider: str
tier: int
latency_class: str
reasoning_mode: str
context_variant: str
effective_context_window: int
# What the scoring actually read: median USD actually billed and median
# gCO2eq over the reference workload. `list_price_per_1m` rides along
# because it is NOT what gets billed, and the gap is worth seeing.
cost: Optional[float] = None
energy: Optional[float] = None
# Still reported — carbon is logged and worth seeing — but it is no longer
# part of the decision.
eco: Optional[float] = None
list_price_per_1m: Optional[float] = None
composite: float
cost_score: float
proficiency_score: float
class RouteResponse(BaseModel):
classification: Classification
latency_tolerance: str
candidates_considered: int
selected: Optional[Candidate]
runners_up: list[Candidate] = []
class DispatchResponse(BaseModel):
route: RouteResponse
content: str
prompt_tokens: Optional[int]
completion_tokens: Optional[int]
telemetry: "Telemetry"
# --- infrastructure -------------------------------------------------------
def _ms(started: float) -> int:
"""Elapsed milliseconds since a perf_counter mark, for the latency fields.
Integers because nobody debugging a route cares about microseconds, and a
log line reads better without six decimal places on every timing.
"""
return int((time.perf_counter() - started) * 1000)
def _db() -> sqlite3.Connection:
conn = sqlite3.connect(cfg.database.path)
conn.row_factory = sqlite3.Row
return conn
def ensure_route_decisions(conn: sqlite3.Connection) -> None:
"""Idempotently create the route_decisions observability table.
schema.sql is CREATE TABLE IF NOT EXISTS, so it defines a NEW database and
silently does nothing to an existing one — the same reason
proficiency_store.ensure_columns exists. A live router.db predating this
table therefore never gets it from re-running schema.sql, so the table is
created here, from code, with a guard. This is the ONLY way it appears on
a live database; the live DB is never recreated or dropped.
Safe to call any number of times against an existing connection: the
CREATE TABLE and CREATE INDEX are both IF NOT EXISTS, so a table that is
already present is left fully intact (rows included) and a second call
no-ops.
"""
conn.execute(
"""
CREATE TABLE IF NOT EXISTS route_decisions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
observed_at TEXT NOT NULL,
kind TEXT NOT NULL,
task_category TEXT,
task_tier INTEGER,
required_context_tokens INTEGER,
confidence REAL,
classifier_ms INTEGER,
classification_source TEXT,
latency_tolerance TEXT,
candidates_considered INTEGER,
selected_model TEXT,
selected_provider TEXT,
runner_up_models TEXT,
est_cost_usd REAL,
est_proficiency REAL,
rejected_reason TEXT,
session_key TEXT,
tools INTEGER,
images INTEGER,
json_mode INTEGER,
streamed INTEGER
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
"ON route_decisions (observed_at)"
)
conn.commit()
def _ensure_route_decisions_table() -> None:
"""Create route_decisions on the live DB if it predates this feature."""
try:
conn = sqlite3.connect(cfg.database.path)
try:
ensure_route_decisions(conn)
finally:
conn.close()
except Exception as exc: # noqa: BLE001
logs.warning(
"startup_route_decisions_migration_failed", error=str(exc)
)
_ensure_route_decisions_table()
def _classifier_client() -> OpenAI:
"""The classification endpoint, which need not be local.
Classification is the one blocking LLM call on the request path, and it
assumes the least about its host: it is a short prompt with a short JSON
answer. That makes it the piece worth moving when the machine running this
router has no GPU worth the name — an RPi or a laptop can route perfectly
well while something else does the classifying.
``api_key_env`` unset means unauthenticated, which is the local Ollama
case: it ignores the key but the SDK requires one to be set.
max_retries=0 matters: the SDK retries twice by default, so
classifier.timeout_seconds silently becomes a 3x wall-clock bound and a
cold model load can block a caller for six minutes on a 120s setting —
observed as a request that hung past 250s and logged nothing. Retrying a
local model that is busy loading only queues more work behind it, so the
configured timeout should be the real one.
"""
key_env = cfg.classifier.api_key_env
if key_env:
api_key = os.environ.get(key_env)
if not api_key:
# Raised rather than passed through as a placeholder: a remote
# classifier with no key 401s on every request, and classify()
# would swallow that into a silent fallback tier forever.
raise HTTPException(
503,
f"classifier.api_key_env={key_env!r} is set but {key_env} is "
"not in the environment.",
)
else:
api_key = "ollama"
return OpenAI(
base_url=cfg.classifier.base_url, api_key=api_key, max_retries=0
)
def _provider_client(provider: str) -> OpenAI:
try:
settings = cfg.dispatch_providers[provider]
except KeyError:
raise HTTPException(500, f"No dispatch config for provider {provider!r}")
api_key = os.environ.get(settings.api_key_env)
if not api_key:
raise HTTPException(
503,
f"{settings.api_key_env} is not set — copy .env.example to .env and fill it in.",
)
return OpenAI(base_url=settings.base_url, api_key=api_key)
# --- classification -------------------------------------------------------
def clamp_for_classifier(text: str, max_chars: int) -> str:
"""Shrink classifier input to the part that carries the instruction.
The classifier decides a category and a tier. It does not need the
document, only the instruction wrapped around it — and feeding it the
document is actively harmful, not merely wasteful. Measured on a ~20k-token
prompt, both local models failed and neither failed cleanly:
| model | wall clock | result |
|---|---|---|
| qwen3.5 | 28.7s | `finish_reason=length`, empty content |
| mistral-nemo | 41.8s | echoed the input back inside the JSON |
Both end as unparseable output, i.e. ~30-40s of local inference spent to
reach `source: "fallback"` — the same answer an immediate failure would
have given, minus the wait.
Head AND tail, because the instruction sits at either end depending on how
the caller phrased it ("Translate this: <doc>" vs "<doc> — translate
this"), and the middle of a pasted document is the one part that never
carries it.
Nothing is lost by clamping. ``required_context_tokens`` does not come
from this call in the path that matters — ``chat_completions`` measures
the real conversation with ``estimate_prompt_tokens`` and takes the
larger, precisely because the classifier's own estimate was measured two
orders of magnitude low.
"""
if max_chars <= 0 or len(text) <= max_chars:
return text
half = max_chars // 2
return f"{text[:half]}\n\n[... {len(text) - max_chars} characters elided ...]\n\n{text[-half:]}"
def classify(task: str, context: Optional[str]) -> Classification:
"""Ask the local model to categorize and size the task.
The allowed categories are appended from config rather than duplicated in
the prompt text, so the classifier cannot return a label that joins
against nothing in the proficiency table.
"""
categories = cfg.proficiency.categories
system_prompt = (
f"{cfg.classifier.system_prompt}\n"
f"Allowed values for task_category (use EXACTLY one of these strings):\n"
+ "\n".join(f" - {c}" for c in categories)
)
# _classifier_user_content handles None context internally (returns task),
# so the outer if/else is redundant — removed for clarity.
user_content = _classifier_user_content(
task, context, cfg.classifier.context_framing
)
user_content = clamp_for_classifier(user_content, cfg.classifier.max_input_chars)
client = _classifier_client()
started = time.perf_counter()
try:
resp = _classify_once(client, system_prompt, user_content)
except (OpenAIError, ValueError, KeyError, TypeError, json.JSONDecodeError) as e:
# A local model that is slow, restarting, or mid-thought must not take
# the caller down with it. Routing degrades to a configured mid tier,
# flagged as source='fallback' so the response says so.
logs.warning(
"fallback",
error=type(e).__name__,
detail=str(e)[:200],
tier=cfg.classifier.fallback_tier,
cat=cfg.classifier.fallback_category,
ms=_ms(started),
)
return Classification(
task_category=cfg.classifier.fallback_category,
task_tier=cfg.classifier.fallback_tier,
required_context_tokens=0,
confidence=0.0,
source="fallback",
)
# The single blocking LLM call on the request path, so its latency is the
# floor under every routed request and worth seeing on its own.
logs.debug(
"classify",
cat=resp.task_category,
tier=resp.task_tier,
ctx=resp.required_context_tokens,
confidence=resp.confidence,
chars=len(user_content),
ms=_ms(started),
)
return resp
def _classifier_user_content(
task: str, context: Optional[str], framing: bool
) -> str:
"""Assemble the classifier's user message (pure, for testability).
``framing`` selects llmrouter's "Context: <prev> / Message: <current>"
layout so a short follow-up inherits its prior turn's complexity;
otherwise the legacy "task / --- context --- / context" layout is used.
In both layouts the framing instruction is appended when context is
present, telling the classifier to treat follow-ups by the prior
complexity.
"""
if not context:
return task
if framing:
user_content = (
f"Context: {context}\n---\nMessage: {task}"
)
else:
user_content = f"{task}\n\n--- context ---\n{context}"
user_content += (
"\n\nA short follow-up message ('Yes', 'Try now?', 'Go ahead') "
"continues the prior turn, so classify by the CONTEXT's "
"complexity, not in isolation."
)
return user_content
def _classify_once(client: OpenAI, system_prompt: str, user_content: str) -> Classification:
"""One classifier round-trip. Raises on anything unusable."""
categories = cfg.proficiency.categories
resp = client.chat.completions.create(
model=cfg.classifier.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content},
],
response_format={"type": "json_object"}
if cfg.classifier.response_format == "json"
else None,
temperature=cfg.classifier.temperature,
max_tokens=cfg.classifier.max_output_tokens,
timeout=cfg.classifier.timeout_seconds,
)
raw = resp.choices[0].message.content or ""
parsed = json.loads(raw)
category = parsed.get("task_category")
if category not in categories:
# A label outside the configured set would silently miss every
# proficiency row, so fall back rather than routing on a phantom join.
category = "general_chat" if "general_chat" in categories else categories[0]
return Classification(
task_category=category,
task_tier=int(parsed["task_tier"]),
required_context_tokens=int(parsed["required_context_tokens"]),
confidence=float(parsed["confidence"]),
)
def apply_escalation(c: Classification) -> Classification:
"""Bump the tier when the classifier isn't confident in its own call."""
if not cfg.escalation.enabled:
return c
# A fallback tier is a deliberate default, not a shaky guess — escalating
# it would silently send every request to the frontier tier whenever the
# local model is unavailable, which is the expensive failure mode.
if c.source == "fallback":
return c
# Off by default. Bumping on uncertainty pays more before anything has
# gone wrong; the iteration budget spends after a check has actually
# failed, which is better on both mandates.
if not cfg.escalation.preemptive_on_low_confidence:
return c
if c.confidence >= cfg.escalation.min_confidence_before_bump:
return c
bumped = min(c.task_tier + 1, cfg.escalation.max_tier)
if bumped == c.task_tier:
return c
return c.model_copy(update={"task_tier": bumped, "escalated": True})
# --- candidate lookup -----------------------------------------------------
def load_candidates(conn: sqlite3.Connection, category: str) -> list[dict]:
"""All model rows, joined to this category's proficiency and mean carbon.
Filtering happens in ``routing.py`` rather than in SQL so the hard filters
stay in one testable place; the catalog is 19 rows.
``eco`` and ``energy`` come ONLY from the fixed reference workload written
by ``seed_energy.py`` (``task_category='seed_reference'``). ``cost`` is
read here too, but as a FALLBACK: ``rank_candidates`` overwrites it with
``routing.estimated_cost``, priced from catalog token prices scaled to the
request in hand. A fixed 400-token sweep ranks models backwards for real
traffic -- glm-5.2-fast looked 3.2x cheaper than deepseek-v4-flash on a toy
prompt and is 5.0x more expensive on a 70k one -- so the measured median
survives only for a row the catalog has no price for.
Reference-only for the rest, because energy depends far more on the shape
of a request than on the model: across organic traffic on a single model,
per-request energy spanned 19x purely because prompt and completion
sizes differed. Averaging that in would rank models by what they
happened to be asked. Organic observations stay in the table for cost
accounting; they just don't steer routing.
Both read the provider's own attributed figures — the billed
``cost_usd`` and the reported ``carbon_g_co2eq`` — deliberately.
That attribution (``energy_kwh = avg_power_watts x duration_seconds x
attribution_ratio``, where the ratio is the request's share of a shared
GPU pool) looks like noise up close: eight rapid identical calls varied
20x, correlating +0.997 with the ratio. It is not. Across the reference
sweep the median ratio spans **750x between models** while the typical
within-model spread is **1.8x** — it is a stable per-model property, and
the values are quantized (0.001, 0.25, 0.5, 0.75), which is what serving
concurrency looks like. A model sharing its GPUs with far more concurrent
requests genuinely costs less per request, and that is most of the real
cost difference in the catalog. Scoring on pre-attribution energy was
tried and discards a 750x signal to suppress a 1.8x one.
The MEDIAN, not the mean, handles what noise remains: a few models
(kimi-k3, kimi-k3-fast, qwen3.6-35b-fast) still show ~50-90x within-model
outliers, and with 7 samples the median ignores up to three of them.
Carbon is dropped where ``carbon_source`` is ``static_fallback``. That is
a constant NeuralWatt substitutes when it cannot resolve live grid data,
so scoring on it would rank a model against a placeholder; without it the
model simply has no eco data and scores the neutral 0.5.
Models with no usable reference samples come back None and take that
same neutral 0.5, so a newly listed model is never penalized for not
having been swept yet.
"""
rows = conn.execute(
"""
SELECT m.*,
p.blended_score AS proficiency,
tp.blended_score AS tool_proficiency
FROM models m
LEFT JOIN proficiency p
ON p.model_id = m.model_id
AND p.provider = m.provider
AND p.category = ?
LEFT JOIN proficiency tp
ON tp.model_id = m.model_id
AND tp.provider = m.provider
AND tp.category = ?
""",
(category, cfg.routing.tool_use_category),
).fetchall()
# Medians are computed here rather than in SQL — SQLite has no MEDIAN, and
# the catalog is 19 rows, so clarity beats a window-function expression.
costs: dict[tuple, list[float]] = {}
carbons: dict[tuple, list[float]] = {}
energies: dict[tuple, list[float]] = {}
for o in conn.execute(
"""
SELECT model_id, provider, cost_usd, energy_kwh, carbon_g_co2eq, carbon_source
FROM energy_observations WHERE task_category = ?
""",
(SEED_CATEGORY,),
):
key = (o["model_id"], o["provider"])
if o["cost_usd"] is not None:
costs.setdefault(key, []).append(o["cost_usd"])
# Energy, not just its dollar equivalent: the plan is a fixed kWh
# quota, so energy is the resource the ceiling has to bound.
if o["energy_kwh"] is not None:
energies.setdefault(key, []).append(o["energy_kwh"])
if (
o["carbon_g_co2eq"] is not None
and o["carbon_source"] != FALLBACK_CARBON_SOURCE
):
carbons.setdefault(key, []).append(o["carbon_g_co2eq"])
out = []
for r in rows:
row = dict(r)
key = (row["model_id"], row["provider"])
row["cost"] = median(costs[key]) if key in costs else None
row["energy"] = median(energies[key]) if key in energies else None
row["eco"] = median(carbons[key]) if key in carbons else None
row["samples"] = len(costs.get(key, []))
out.append(row)
return out
def model_output_ceiling(model_id: str, provider: str) -> Optional[int]:
"""The model's own advertised output limit, if the catalog states one.
Bounds a truncation retry: doubling the budget past what the model accepts
would be rejected outright, wasting the attempt.
"""
conn = _db()
try:
row = conn.execute(
"SELECT max_output_tokens FROM models WHERE model_id = ? AND provider = ?",
(model_id, provider),
).fetchone()
finally:
conn.close()
return row["max_output_tokens"] if row else None
def _to_candidate(row: dict) -> Candidate:
fields = {k: row[k] for k in Candidate.model_fields if k in row}
fields["list_price_per_1m"] = row.get("cost_per_1m_completion")
return Candidate(**fields)
def route(req: TaskRequest) -> RouteResponse:
latency_tolerance = req.latency_tolerance or cfg.routing.default_latency_tolerance
if req.task_category and req.task_tier and req.required_context_tokens is not None:
classification = Classification(
task_category=req.task_category,
task_tier=req.task_tier,
required_context_tokens=req.required_context_tokens,
confidence=1.0,
source="override",
)
else:
classification = apply_escalation(classify(req.task, req.context))
if req.task_category:
classification = classification.model_copy(
update={"task_category": req.task_category}
)
if req.task_tier:
classification = classification.model_copy(update={"task_tier": req.task_tier})
if req.required_context_tokens is not None:
classification = classification.model_copy(
update={"required_context_tokens": req.required_context_tokens}
)
conn = _db()
try:
rows = load_candidates(conn, classification.task_category)
finally:
conn.close()
filters = dict(
required_context_tokens=classification.required_context_tokens,
required_tier=classification.task_tier,
latency_tolerance=latency_tolerance,
allowed_access_levels=cfg.routing.allowed_access_levels,
exclude_stale=cfg.freshness.exclude_stale,
exclude_deprecated=cfg.freshness.exclude_deprecated,
min_tool_proficiency=(
cfg.routing.min_tool_proficiency if req.tools_present else None
),
require_vision=cfg.routing.require_vision if req.has_images else False,
require_json_mode=(
cfg.routing.require_json_mode if req.require_json_mode else False
),
)
eligible = select_candidates(rows, **filters)
if logs.enabled_for_debug():
# "No model satisfies the hard filters" is otherwise a dead end with no
# explanation. One line per drop, naming the filter and its numbers, is
# the difference between a five-minute answer and an afternoon.
survivors = {r["model_id"] for r in eligible}
for row in rows:
if row["model_id"] not in survivors:
logs.debug(
"filter",
model=row["model_id"],
reason=rejection_reason(row, **filters),
)
ranked = rank_candidates(
eligible,
quality_tolerance=cfg.objective.quality_tolerance,
max_energy_per_request=cfg.objective.max_energy_per_request,
# Priced for the request in hand, not for a benchmark.
prompt_tokens=classification.required_context_tokens,
completion_tokens=cfg.objective.assumed_completion_tokens,
cache_rate=cfg.objective.assumed_cache_rate,
)
if logs.enabled_for_debug():
for position, row in enumerate(ranked[:5]):
logs.debug(
"rank",
pos=position,
model=row["model_id"],
prof=row["proficiency_score"],
est_usd=row.get("cost"),
kwh=row.get("energy"),
)
return RouteResponse(
classification=classification,
latency_tolerance=latency_tolerance,
candidates_considered=len(eligible),
selected=_to_candidate(ranked[0]) if ranked else None,
runners_up=[_to_candidate(r) for r in ranked[1:4]],
)
def log_decision(
decision: RouteResponse,
*,
tools: bool,
ms: int,
src: Optional[str] = None,
ctx_src: str = "classifier",
) -> None:
"""The one INFO line per request that says what the router decided.
Emitted by the ENDPOINTS rather than by ``route()``, because
``chat_completions`` routes twice when the measured conversation turns out
larger than the classifier's estimate -- which, with opencode sending ~32K
of system prompt, is essentially every request. Logging inside route()
would double every line for no new information.
That second pass is also why ``src`` can be overridden. The re-route passes
category and tier back in, so the resulting Classification reads
``source='override'`` -- which in a log line means "the client chose this",
and the client did no such thing: the classifier ran, and only the context
figure was replaced. ``ctx_src`` records that separately, so the line says
what actually happened rather than what the second call looked like.
"""
selected = decision.selected
logs.info(
"route",
cat=decision.classification.task_category,
tier=decision.classification.task_tier,
ctx=decision.classification.required_context_tokens,
ctx_src=ctx_src,
src=src or decision.classification.source,
tools=tools,
latency=decision.latency_tolerance,
cand=decision.candidates_considered,
pick=selected.model_id if selected else None,
est_usd=selected.cost if selected else None,
prof=selected.proficiency_score if selected else None,
ms=ms,
)
def persist_route_decision(
decision_kind: str,
*,
classification=None,
latency_tolerance=None,
selected_model=None,
selected_provider=None,
runners_up=None,
rejected_reason=None,
session_key=None,
tools=0,
images=0,
json_mode=0,
streamed=0,
classification_source=None,
classifier_ms=None,
) -> None:
"""Record one routing decision to route_decisions, best-effort and gated.
Every decision path calls this exactly once per request. A write that
fails for any reason (read-only or locked DB, missing table) is logged at
warning and swallowed, because a decision record is worth having but never
worth failing or slowing a request for. Gated by
``cfg.logging.log_route_decisions`` so the monitoring history can be left
empty where it is not wanted.
``classification`` may be a RouteResponse (deriving category/tier/context/…
from its Classification and the selected model from its ``selected``) or a
bare Classification; the explicit arguments cover the paths with no
RouteResponse (passthrough, local-vision) and let a caller override what
the RouteResponse would otherwise say — the chat re-route's
Classification reads ``source='override'`` even though the classifier did
decide the request, so the caller passes the first route's actual source.
"""
# The gate is read before anything can fail: turning the table off must be
# a guaranteed no-op even on a broken DB.
if not cfg.logging.log_route_decisions:
return
clf = None
selected = None
derived_runners = None
candidates = None
derived_latency = latency_tolerance
if isinstance(classification, RouteResponse):
clf = classification.classification
selected = classification.selected
derived_runners = classification.runners_up
candidates = classification.candidates_considered
derived_latency = classification.latency_tolerance
elif isinstance(classification, Classification):
clf = classification
if clf is not None and classification_source is None:
classification_source = clf.source
# An override-created Classification never consulted the classifier, so it
# has no classifier latency worth recording — even when a caller passed a
# timing value (route_endpoint's whole-route ms would otherwise leak in).
if classification_source == "override":
classifier_ms = None
if selected is not None:
model = selected.model_id
provider = selected.provider
est_cost = selected.cost
est_prof = selected.proficiency_score
runner_json = (
json.dumps(
[{"model_id": c.model_id, "provider": c.provider}
for c in derived_runners[:3]]
)
if derived_runners
else None
)
else:
model = selected_model
provider = selected_provider
est_cost = None
est_prof = None
runner_json = json.dumps(runners_up[:3]) if runners_up else None
conn = _db()
try:
# Guarantee the table on the WRITE path, mirroring
# proficiency_store._write -> ensure_columns. schema.sql is CREATE
# TABLE IF NOT EXISTS so it never adds the table to a live router.db
# that predates it; the module-load hook covers the normal boot, but
# any other path that calls persist first (a test harness, a lazy
# import, a future refactor) must still get the table here or the
# INSERT raises no-such-table and the decision is silently lost.
ensure_route_decisions(conn)
observed_at = datetime.now(timezone.utc).isoformat()
cursor = conn.execute(
"""
INSERT INTO route_decisions (
observed_at, kind, task_category, task_tier,
required_context_tokens, confidence, classifier_ms,
classification_source, latency_tolerance, candidates_considered,
selected_model, selected_provider, runner_up_models,
est_cost_usd, est_proficiency, rejected_reason, session_key,
tools, images, json_mode, streamed
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
observed_at,
decision_kind,
clf.task_category if clf is not None else None,
clf.task_tier if clf is not None else None,
clf.required_context_tokens if clf is not None else None,
clf.confidence if clf is not None else None,
classifier_ms,
classification_source,
derived_latency,
candidates,
model,
provider,
runner_json,
est_cost,
est_prof,
rejected_reason,
session_key,
int(bool(tools)),
int(bool(images)),
int(bool(json_mode)),
int(bool(streamed)),
),
)
decision_id = int(cursor.lastrowid)
conn.commit()
# Fan the recorded decision out to any live dashboard subscribers.
# The row id becomes the ordering handle the TUI keeps and lets a
# detail popup re-target the exact decision the SSE feed reported.
events.publish_decision(
{
"id": decision_id,
"observed_at": observed_at,
"kind": decision_kind,
"task_category": clf.task_category if clf is not None else None,
"task_tier": clf.task_tier if clf is not None else None,
"required_context_tokens": (
clf.required_context_tokens if clf is not None else None
),
"confidence": clf.confidence if clf is not None else None,
"classifier_ms": classifier_ms,
"classification_source": classification_source,
"latency_tolerance": derived_latency,
"candidates_considered": candidates,
"selected_model": model,
"selected_provider": provider,
"runner_up_models": runner_json,
"est_cost_usd": est_cost,
"est_proficiency": est_prof,
"rejected_reason": rejected_reason,
"session_key": session_key,
"tools": int(bool(tools)),
"images": int(bool(images)),
"json_mode": int(bool(json_mode)),
"streamed": int(bool(streamed)),
}
)
except Exception as e: # noqa: BLE001 - best-effort must never raise
logs.warning(
"route_decision_persist",
kind=decision_kind,
error=type(e).__name__,
)
finally:
conn.close()
# --- energy / cost accounting --------------------------------------------
class Telemetry(BaseModel):
"""What the provider reported about a completion it just served."""
energy_kwh: Optional[float] = None
energy_btu: Optional[float] = None
avg_power_watts: Optional[float] = None
duration_seconds: Optional[float] = None
attribution_ratio: Optional[float] = None
carbon_g_co2eq: Optional[float] = None
grid_carbon_intensity: Optional[float] = None
grid_id: Optional[str] = None
carbon_source: Optional[str] = None
cost_usd: Optional[float] = None
allowance_remaining_usd: Optional[float] = None
service_tier: Optional[str] = None
def extract_telemetry(payload: dict) -> Telemetry:
"""Read NeuralWatt's per-request energy and cost blocks.
Both sit at the top level of the completion response, outside anything
the OpenAI schema models, so this reads the raw JSON rather than the
parsed object. Every field is optional: a provider that reports nothing
yields an all-None Telemetry rather than an error, and the neutral-0.5
path in scoring covers the gap.
"""
energy = payload.get("energy") or {}
cost = payload.get("cost") or {}
kwh = energy.get("energy_kwh")
return Telemetry(
energy_kwh=kwh,
energy_btu=kwh * BTU_PER_KWH if kwh is not None else None,
avg_power_watts=energy.get("avg_power_watts"),
duration_seconds=energy.get("duration_seconds"),
attribution_ratio=energy.get("attribution_ratio"),
carbon_g_co2eq=energy.get("carbon_g_co2eq"),
grid_carbon_intensity=energy.get("grid_carbon_intensity_gco2perkwhr"),
grid_id=energy.get("grid_id"),
carbon_source=energy.get("carbon_source"),
# The provider's billed figure, not a tokens x list-price estimate.
# Flex rows bill under their standard sibling despite identical
# advertised pricing, so the estimate is wrong for every flex call.
cost_usd=cost.get("request_cost_usd"),
allowance_remaining_usd=cost.get("allowance_remaining_usd"),
service_tier=payload.get("service_tier"),
)
def log_verification(
model_id: str,
provider: str,
task_category: Optional[str],
request_id: Optional[str] = None,
*,
kind: str,
verdict: str,
detail: str,
completion_tokens: Optional[int] = None,
model_attributable: bool = True,
) -> None:
"""Record what a check said about one completion.
Recorded for every response including 'unverifiable', because the rate at
which responses cannot be checked is itself worth knowing — if most
traffic is unverifiable, structural checking is not earning its place.
"""
conn = _db()
try:
conn.execute(
"""
INSERT INTO verifications (
model_id, provider, request_id, task_category, kind, verdict, detail,
completion_tokens, observed_at, model_attributable
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
model_id,
provider,
request_id,
task_category,
kind,
verdict,
detail[:300] if detail else None,
completion_tokens,
datetime.now(timezone.utc).isoformat(),
int(model_attributable),
),
)
conn.commit()
finally:
conn.close()
def run_local_verification(
model_id: str,
provider: str,
task_category: Optional[str],
request_text: str,
answer: str,
completion_tokens: Optional[int],
request_id: Optional[str] = None,
) -> None:
"""Second-opinion check on an answer nothing structural could judge.
Runs on the LOCAL model, after the response has already gone back to the
client, so its ~6s never lands on anyone's latency. Failures here are
swallowed: a checker that cannot answer must not take down a request that
already succeeded, and must not record a verdict either — no sample beats
a false one.
"""
# Ollama's NATIVE endpoint, not the OpenAI-compatible one, because only it
# exposes `think`. That matters more than the inconsistency of using two
# APIs: with thinking on, this model spent its ENTIRE budget reasoning and
# never emitted the verdict — 2,048 tokens produced 7,880 characters of
# thought and empty content, and raising the cap to 8,192 just bought more
# thought. With think=False it answers in 25 tokens, and returned the same
# verdict on 3/3 repeats of four labelled cases.
#
# This endpoint is configured separately from the classifier's and is NOT
# derived from it. It used to be, and that silently coupled two unrelated
# decisions: pointing classification at a cloud provider would have sent
# these requests to <provider>/api/chat, which does not exist.
url = cfg.verification.base_url.rstrip("/")
if url.endswith("/v1"):
url = url[: -len("/v1")]
verify_model = cfg.verification.model or cfg.classifier.model
try:
resp = requests.post(
f"{url}/api/chat",
json={
"model": verify_model,
"think": False,
"stream": False,
"format": "json",
"options": {
"temperature": 0,
"num_predict": cfg.verification.max_output_tokens,
},
"messages": [
{"role": "system", "content": LOCAL_VERIFY_SYSTEM},
{
"role": "user",
"content": build_local_verify_prompt(request_text, answer),
},
],
},
timeout=cfg.verification.timeout_seconds,
)
resp.raise_for_status()
raw = ((resp.json() or {}).get("message") or {}).get("content") or ""
except (requests.RequestException, ValueError) as e:
logs.warning("verify_unavailable", error=type(e).__name__, model=verify_model)
return
check = interpret_local_verdict(parse_verdict_json(raw))
if check is None:
logs.warning("verify_unusable", model=verify_model)
return
log_verification(
model_id, provider, task_category, request_id,
kind="local_llm", verdict=check.verdict, detail=check.detail,
completion_tokens=completion_tokens,
)
def log_observation(
model_id: str,
provider: str,
task_category: str,
request_id: Optional[str],
session_key: Optional[str] = None,
session_dir: Optional[str] = None,
*,
prompt_tokens: Optional[int],
completion_tokens: Optional[int],
telemetry: Telemetry,
) -> None:
if not cfg.logging.log_energy_observations:
return
conn = _db()
try:
conn.execute(
"""
INSERT INTO energy_observations (
model_id, provider, request_id, session_key, session_dir,
task_category, prompt_tokens, completion_tokens,
energy_kwh, energy_btu, avg_power_watts, duration_seconds,
attribution_ratio, carbon_g_co2eq, grid_carbon_intensity, grid_id,
carbon_source, cost_usd, allowance_remaining_usd, service_tier, observed_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
model_id,
provider,
request_id,
session_key,
session_dir,
task_category,
prompt_tokens,
completion_tokens,
telemetry.energy_kwh,
telemetry.energy_btu,
telemetry.avg_power_watts,
telemetry.duration_seconds,
telemetry.attribution_ratio,
telemetry.carbon_g_co2eq,
telemetry.grid_carbon_intensity,
telemetry.grid_id,
telemetry.carbon_source,
telemetry.cost_usd,
telemetry.allowance_remaining_usd,
telemetry.service_tier,
datetime.now(timezone.utc).isoformat(),
),
)
conn.commit()
finally:
conn.close()
# --- endpoints ------------------------------------------------------------
@app.get("/health")
def health():
conn = _db()
try:
counts = dict(
conn.execute(
"""
SELECT 'models', COUNT(*) FROM models
UNION ALL SELECT 'routable', COUNT(*) FROM models
WHERE access_level = 'public' AND availability = 'active'
UNION ALL SELECT 'proficiency', COUNT(*) FROM proficiency
UNION ALL SELECT 'energy_observations', COUNT(*) FROM energy_observations
"""
).fetchall()
)
scoring = scoring_coverage(conn, cfg)
finally:
conn.close()
try:
_classifier_client().models.list()
classifier_ok = True
except OpenAIError:
classifier_ok = False
return {
"status": "ok",
"counts": counts,
"scoring": scoring,
"classifier_reachable": classifier_ok,
"classifier_model": cfg.classifier.model,
# The tier labels, documented as being for logging and dashboards.
# /health is the dashboard surface, so this is where they reach one.
"tiers": cfg.tiers,
"providers": list(cfg.dispatch_providers),
"api_keys_present": {
name: bool(os.environ.get(p.api_key_env))
for name, p in cfg.dispatch_providers.items()
},
}
def _quota_burn(conn: "sqlite3.Connection") -> Optional[dict]:
"""Thin wrapper so /health can call quota_burn with the SAME connection."""
return quota_burn(conn, cfg)
@app.get("/metrics")
def metrics_endpoint():
"""Aggregated read-only view over the router's observability tables.
Mirrors the metrics.py helpers into one JSON payload for a dashboard.
Unauthenticated and loopback-bound exactly like /health — it contains no
conversation text, prompt, or session_dir.
"""
conn = _db()
try:
# per_model aggregates only energy_observations (cloud completions);
# local-vision fallback decisions appear in recent_decisions but not
# per_model (no cloud call), so the two views legitimately differ.
return {
"quota": quota_burn(conn, cfg),
"coverage": scoring_coverage(conn, cfg),
"recent_decisions": recent_decisions(conn),
"per_model": per_model(conn),
"verdict_mix": verdict_mix(conn),
"top_proficiency": top_proficiency(conn, "coding_general"),
"generated_at": datetime.now(timezone.utc).isoformat(),
}
finally:
conn.close()
# SSE keep-alive cadence and reconnect hint, in seconds.
SSE_HEARTBEAT_SECONDS = 15
SSE_RETRY_SECONDS = 3
def _sse_data(payload: dict) -> bytes:
"""Format a dict as a SSE ``data:`` frame (bytes)."""
return f"data: {json.dumps(payload)}\n\n".encode()
def _drain_queue(subscriber: queue.Queue) -> list[dict[str, Any]]:
"""Drain a subscriber queue without blocking; stops at the eviction sentinel.
Returns what was waiting (excluding the sentinel itself).
"""
drained: list[dict[str, Any]] = []
while True:
try:
item = subscriber.get_nowait()
except queue.Empty:
return drained
if item is events._EVICTED:
return drained
drained.append(item)
def _decision_event_stream():
"""Yield route decisions as SSE ``data:`` frames, then live ones.
Replays the broker's recent ring-buffer contents first so a new dashboard
connection immediately sees recent history, then blocks on the subscriber
queue for new decisions. A ``:heartbeat`` comment is emitted on idle so
middleboxes do not drop the connection, and the subscriber is always
unsubscribed on exit.
When a subscriber is evicted (its queue became full), the broker writes
a sentinel value to the queue. This generator detects the sentinel, breaks
out of the loop, and lets ``finally`` unsubscribe, closing the SSE
connection cleanly so the TUI can reconnect.
"""
subscriber = events.subscribe(replay=True)
try:
yield f"retry: {int(SSE_RETRY_SECONDS * 1000)}\n\n"
for decision in _drain_queue(subscriber):
yield _sse_data(decision)
while True:
try:
decision = subscriber.get(timeout=SSE_HEARTBEAT_SECONDS)
except queue.Empty:
yield ":heartbeat\n\n"
continue
if decision is events._EVICTED:
break
yield _sse_data(decision)
finally:
events.unsubscribe(subscriber)
@app.get("/events/decisions")
async def events_decisions():
"""Server-sent-events stream of routing decisions for live dashboards.
The first frames replay recent decisions (so the TUI can populate its
table before any new traffic), then new decisions stream as they are
recorded. Unauthenticated and loopback-bound like ``/metrics``; each
event carries only the fields already on a ``route_decisions`` row — no
conversation text, prompt, or session_dir.
"""
return StreamingResponse(
_decision_event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
class OutcomeReport(BaseModel):
"""A client telling the router whether an answer actually worked."""
request_id: Optional[str] = Field(
None,
description=(
"The completion id the router returned (`id` in the response body, "
"and on every stream chunk). Omit it and the report attaches to the "
"most recent completion this router served — see the endpoint docs "
"for when that is safe."
),
)
ok: bool = Field(..., description="Did the answer actually do the job?")
detail: Optional[str] = Field(
None, description="Optional short note — an error message, what broke."
)
source: Optional[str] = Field(
None, description="Where the report came from, e.g. the project directory."
)
AMBIGUOUS = object()
def _most_recent_if_unambiguous(conn: sqlite3.Connection):
"""The latest completion, but only when one conversation is active.
Guessing here is how quality data gets corrupted: attribute a failing test
run to the wrong session and a model is penalized for work it never did.
Corrupt evidence is worse than missing evidence — this project has already
recorded false failures twice from harness bugs, and both took measurement
to catch.
So when two conversations have been served inside the window, this returns
AMBIGUOUS and the caller refuses rather than picks.
The window is short on purpose. A test run follows the completion that
caused it within seconds; a wide window sweeps in sessions that finished
long ago and makes every report look ambiguous. Observed directly: at 30
minutes, a real opencode run was refused because of test traffic from
earlier in the same session of work.
``julianday()`` on both sides, NOT a string comparison against
``datetime('now', ...)``. ``observed_at`` is written by
``datetime.now(timezone.utc).isoformat()``, so it carries a 'T' separator
(``2026-08-23T00:20:04.577131+00:00``) while ``datetime()`` returns a space
(``2026-08-23 00:18:04``). 'T' sorts after ' ', so once the dates match the
time of day never participates and the window silently becomes "everything
today". Measured on the live DB: 14 rows across 2 sessions matched where
the correct comparison matched 0 -- which is what produced the spurious
409s, and made this setting inert at every value. ``poller.mark_stale``
already had the right form.
"""
window = cfg.verification.outcome_attribution_window_seconds
recent = conn.execute(
f"""
SELECT id, request_id, model_id, provider, task_category, session_key
FROM energy_observations
WHERE request_id IS NOT NULL
AND task_category != ?
AND julianday(observed_at) > julianday('now', '-{int(window)} seconds')
ORDER BY id DESC LIMIT 50
""",
(SEED_CATEGORY,),
).fetchall()
if not recent:
return None
keys = {r["session_key"] for r in recent if r["session_key"]}
if len(keys) > 1:
return AMBIGUOUS
return recent[0]
@app.post("/outcome")
def report_outcome(report: OutcomeReport):
"""Record whether a completion actually worked.
Logged either way. A report that lands is the only ground-truth signal the
router gets; one that is refused is the client discovering its reports go
nowhere, which is worth seeing from this side too.
This is the only ground truth the router can get. Everything else it
records is a proxy: structural checks know whether code *parses*, the
local checker guesses whether prose *looks* right, and neither knows
whether the answer did the job. The client does — it ran the tests, or
used the answer, or watched it fail.
It is also the only quality signal that survives streaming. Retries cannot
reach a streamed response because the bytes are already gone, but a report
arrives afterwards and works the same either way — which matters because
every agent client streams.
Unlike a structural pass, a client-reported SUCCESS is worth recording.
'ok' from a parser means the code parsed, which is weak evidence and would
inflate scores if counted. 'succeeded' from a client means the work
worked.
"""
logs.new_trace()
conn = _db()
try:
if report.request_id:
row = conn.execute(
"""
SELECT id, request_id, model_id, provider, task_category
FROM energy_observations
WHERE request_id = ? ORDER BY id DESC LIMIT 1
""",
(report.request_id,),
).fetchone()
elif report.source:
# The client told us where it is. If any completion came from a
# conversation naming that directory, this is exact even with
# several sessions running.
row = conn.execute(
"""
SELECT id, request_id, model_id, provider, task_category
FROM energy_observations
WHERE session_dir = ? AND request_id IS NOT NULL
AND task_category != ?
ORDER BY id DESC LIMIT 1
""",
(report.source, SEED_CATEGORY),
).fetchone()
if row is None:
row = _most_recent_if_unambiguous(conn)
else:
row = _most_recent_if_unambiguous(conn)
finally:
conn.close()
if row is AMBIGUOUS:
logs.warning(
"outcome_refused",
reason="ambiguous",
source=report.source,
ok=report.ok,
)
raise HTTPException(
409,
"More than one conversation has been routed recently, so this "
"report cannot be attributed with confidence. Pass request_id (the "
"`id` on the completion or any stream chunk) to disambiguate. "
"Refusing rather than guessing: a misattributed outcome penalizes a "
"model for work it never did.",
)
if row is None:
logs.warning(
"outcome_refused",
reason="unknown_request",
rid=report.request_id,
source=report.source,
)
# Deliberately a 404 rather than a silent accept: a client whose
# reports go nowhere should find out, not quietly train nothing.
raise HTTPException(
404,
f"No routed completion found for request_id {report.request_id!r}. "
"Only completions this router dispatched can be reported on."
if report.request_id
else "No completions have been routed yet, so there is nothing to "
"report on.",
)
log_verification(
row["model_id"],
row["provider"],
row["task_category"],
row["request_id"],
kind="client_outcome",
verdict="succeeded" if report.ok else "failed",
detail=" | ".join(x for x in (report.detail, report.source) if x)[:300],
)
logs.info(
"outcome",
verdict="succeeded" if report.ok else "failed",
model=row["model_id"],
rid=row["request_id"],
cat=row["task_category"],
source=report.source,
)
return {
"recorded": True,
"request_id": row["request_id"],
"model_id": row["model_id"],
"task_category": row["task_category"],
"verdict": "succeeded" if report.ok else "failed",
}
@app.post("/route", response_model=RouteResponse)
def route_endpoint(req: TaskRequest):
"""Classify and pick a model without calling it."""
logs.new_trace()
started = time.perf_counter()
decision = route(req)
log_decision(
decision,
tools=req.tools_present,
ms=_ms(started),
ctx_src="caller" if req.required_context_tokens is not None else "classifier",
)
persist_route_decision(
"route",
classification=decision,
tools=req.tools_present,
images=req.has_images,
json_mode=req.require_json_mode,
classifier_ms=_ms(started),
)
return decision
# --- OpenAI-compatible surface -------------------------------------------
# Absolute paths that look like a project root, for pulling a working
# directory out of an agent's system prompt.
CWD_RE = re.compile(r"(/(?:home|Users|tmp|opt|srv|var)/[\w.@-]+(?:/[\w.@-]+)*)")
def session_fingerprint(messages: list[dict]) -> Optional[str]:
"""A stable id for the conversation this request belongs to.
Hashes the opening message, which an agent client holds constant for the
life of a session (it is the system prompt) and which differs between
sessions. That lets the router notice when two clients are talking to it
at once without either of them saying so.
"""
for m in messages:
content = m.get("content")
if isinstance(content, list):
content = " ".join(
p.get("text", "") for p in content if isinstance(p, dict)
)
if isinstance(content, str) and content.strip():
return hashlib.sha256(content[:4000].encode()).hexdigest()[:16]
return None
def session_directory(messages: list[dict]) -> Optional[str]:
"""The working directory this conversation is about, if it reveals one.
Scans the WHOLE conversation, not just the system prompt. opencode does not
state its project root up front — verified by capturing a real request —
but a coding agent names files inside its project constantly, in tool
calls and results. The directory containing the most of those paths is the
working directory.
This is what makes outcome attribution exact under concurrent sessions: a
report carries its own directory, and matching it beats guessing which
conversation was most recent.
"""
counts: dict[str, int] = {}
for m in messages[-40:]:
content = m.get("content")
if isinstance(content, list):
content = " ".join(
p.get("text", "") for p in content if isinstance(p, dict)
)
if not isinstance(content, str):
continue
for path in CWD_RE.findall(content[:4000]):
parts = path.split("/")
# Drop a trailing filename: a path ending in something with an
# extension names a file, and its directory is what a client
# reports as its working directory.
if "." in parts[-1]:
parts = parts[:-1]
# Count every ancestor, so the shared project root accumulates the
# most hits even when each file sits in a different subdirectory.
for depth in range(3, len(parts) + 1):
counts["/".join(parts[:depth])] = counts.get("/".join(parts[:depth]), 0) + 1
if not counts:
return None
# Deepest directory that still explains most of the paths seen. Ties go to
# the longer path so sibling projects under one parent stay distinct.
best = max(counts.values())
threshold = max(2, best * 0.6)
winners = [p for p, n in counts.items() if n >= threshold]
return max(winners, key=len) if winners else None
def estimate_prompt_tokens(
messages: list[dict], tools: Optional[list[dict]] = None
) -> int:
"""Floor estimate of a conversation's token count, from its characters.
Deliberately crude. Its only job is to stop a long conversation being
routed to a model that cannot hold it when the classifier lowballs its
own input size. Tool definitions ride along in ``upstream_body`` on every
call, so they count against the same context window and belong in this
estimate too -- omitting them undercounts every tool-carrying request,
which is nearly all agent traffic.
"""
chars = 0
for m in messages:
content = m.get("content")
if isinstance(content, str):
chars += len(content)
elif isinstance(content, list):
# Multimodal content parts; count the text ones.
for part in content:
if isinstance(part, dict) and isinstance(part.get("text"), str):
chars += len(part["text"])
if tools:
chars += len(json.dumps(tools))
return chars // CHARS_PER_TOKEN
def _last_user_text(messages: list[dict]) -> str:
for m in reversed(messages):
if m.get("role") == "user":
content = m.get("content")
if isinstance(content, str):
return content
if isinstance(content, list):
return " ".join(
p["text"]
for p in content
if isinstance(p, dict) and isinstance(p.get("text"), str)
)
return ""
def _previous_context(messages: list[dict]) -> str:
"""The preceding assistant turn of the last user message.
Walks backwards from the message before the last user turn, returning the
text content of the nearest ``assistant`` message. Skips ``system``,
``user``, and ``tool`` roles so that tool output or the system prompt
never contaminates the framing signal. Uses ``context_prune.extract_text``
for the actual content-block parsing (same logic as the former inlined
block parser, but deduplicated). Returns the concatenated text of the
first assistant message found, truncated to 200 characters, or ``""``
when no such message exists.
"""
for i in range(len(messages) - 2, -1, -1):
msg = messages[i]
if msg.get("role") != "assistant":
continue
text = extract_text(msg)
if text:
return text[:200]
return ""
def _count_images(messages: list[dict]) -> int:
"""The number of image_url parts across the whole conversation.
Scans all messages, not just the last: any image in an earlier turn still
has to be understood by whoever answers the latest one, exactly as in
``detect_capabilities``. A part counts only when it is a dict whose type
is ``image_url``, mirroring the detection rule.
"""
return sum(1 for _ in iter_image_url_values(messages))
def _image_payload_bytes(messages: list[dict]) -> int:
"""Total base64 payload length across all image_url parts.
The "length" is the raw length of the payload string carried by each part,
whatever its shape. A part may spell the data inline in ``image_url.url``
or in a separate ``detail``/``data`` field; this counts the bytes of the
image_url value so an oversized upload is skipped rather than POSTed to a
local model that would reject it.
"""
return sum(len(v) for v in iter_image_url_values(messages))
def _local_vision_data_uris_ok(messages: list[dict]) -> bool:
"""Whether every image_url part is an inline ``data:`` URI, not a URL.
The local vision model (typically Ollama on this host) is trusted to read
the pixels of an inline image, but pointing it at an arbitrary ``http`` URL
would let a request make the local model fetch internal or private
resources the caller cannot reach directly — SSRF by indirection. OpenAI's
``image_url`` schema allows either a base64 ``data:`` URI or a remote URL,
so a URL is a real possibility, not a theoretical edge. The fallback is
therefore refused unless EVERY part is a data URI; a single remote/unknown
URL declines the fallback (falls through to the 422) rather than handing a
network-fetching capability to an unauthenticated caller.
"""
return all(v.startswith("data:") for v in iter_image_url_values(messages))
def _run_local_vision(messages: list[dict], lv_cfg) -> Optional[str]:
"""Ask a local vision model to caption the request's images, if it can.
Runs only when routing found no cloud vision candidate and the local
vision path is enabled — the fallback that keeps a multimodal request
answerable without spending a cloud completion on a model that would 400.
Returns None on any failure or budget refusal so the caller falls through
to the 422; a hidden failure must not masquerade as a 200.
The messages passed are the ORIGINAL list that still carries the image_url
parts — the local model reads the pixels, so the parts cannot be stripped.
No remote URL is ever forwarded: every image part must be an inline
``data:`` URI, or the fallback is declined (see ``_local_vision_data_uris_ok``).
"""
if not _local_vision_data_uris_ok(messages):
logs.warning("local_vision_skip", reason="remote_url")
return None
n_images = _count_images(messages)
if n_images > lv_cfg.max_images:
logs.warning(
"local_vision_skip", reason="too_many_images",
n=n_images, max=lv_cfg.max_images,
)
return None
payload_bytes = _image_payload_bytes(messages)
if payload_bytes > lv_cfg.max_image_bytes:
logs.warning(
"local_vision_skip", reason="oversized_image",
bytes=payload_bytes, max=lv_cfg.max_image_bytes,
)
return None
url = f"{lv_cfg.base_url.rstrip('/')}/chat/completions"
headers = {}
if lv_cfg.api_key_env:
key = os.environ.get(lv_cfg.api_key_env)
if not key:
logs.warning("local_vision_skip", reason="missing_api_key")
return None
headers["Authorization"] = f"Bearer {key}"
try:
resp = requests.post(
url,
headers=headers,
json={
"model": lv_cfg.model,
"messages": messages,
"temperature": 0,
"max_tokens": 2048,
"stream": False,
},
timeout=lv_cfg.timeout_seconds,
)
except requests.RequestException as e:
logs.warning("local_vision_skip", reason="request", error=type(e).__name__)
return None
if resp.status_code != 200:
# Deliberately no response body in the log: an upstream error can echo
# request content or headers back, and the journal must not become a
# place reflected prompt/image data lands. Status alone is enough to
# see that the fallback could not answer.
logs.warning(
"local_vision_skip", reason="status", status=resp.status_code,
)
return None
try:
payload = resp.json()
except ValueError:
logs.warning("local_vision_skip", reason="unparseable")
return None
content = (payload.get("choices") or [{}])[0].get("message", {}).get("content")
# A 200 with empty or non-string content is a failed answer, not a success.
# Returning ``None`` lets the caller fall through to the 422 so an empty
# local caption cannot masquerade as a 200 — the same "no hidden failures"
# contract as every other early return above.
if not isinstance(content, str) or not content.strip():
return None
return content
def _local_vision_response(content: str, *, streaming: bool) -> Response:
"""Shape a local vision answer into an OpenAI completion, streamed or not.
``streaming`` diverges where the delivery differs: buffered answers get the
structural-verification header so a client can see the verdict, streamed
answers cannot carry headers after the first chunk so they just emit the
SSE events.
"""
payload = {
"id": f"local-vision-{secrets.token_hex(6)}",
"object": "chat.completion",
"created": int(time.time()),
"model": cfg.local_vision.model,
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": content},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 0, "completion_tokens": 0},
}
logs.info(
"dispatch",
model=cfg.local_vision.model,
provider="local",
category="general_chat",
stream=streaming,
)
if not streaming:
verdict = verify_response(content, "stop", has_tool_calls=False).verdict
return JSONResponse(
content=payload,
headers={
"X-Router-Model": cfg.local_vision.model,
"X-Router-Verification": verdict,
},
)
def stream():
yield f"data: {json.dumps({'id': payload['id'], 'object': 'chat.completion.chunk', 'created': payload['created'], 'model': payload['model'], 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': content}, 'finish_reason': None}]})}\n\n".encode()
yield f"data: {json.dumps({'id': payload['id'], 'object': 'chat.completion.chunk', 'created': payload['created'], 'model': payload['model'], 'choices': [{'index': 0, 'delta': {}, 'finish_reason': 'stop'}], 'usage': payload['usage']})}\n\n".encode()
yield b"data: [DONE]\n\n"
return StreamingResponse(stream(), media_type="text/event-stream")
def _model_exists(model_id: str) -> bool:
"""Whether a bare id names a real routable catalog row.
Used to decide whether a `provider/model` string a client sent is the
client's own provider alias (opencode's `llm-router/...` convention,
never a real NeuralWatt id) and should be stripped, or a real id that
happens to contain a slash -- which would not resolve once stripped.
"""
conn = _db()
try:
row = conn.execute(
"SELECT 1 FROM models WHERE model_id = ? AND provider = 'neuralwatt'",
(model_id,),
).fetchone()
finally:
conn.close()
return row is not None
def _check_pinned_capabilities(model_id: str, caps) -> None:
"""Reject a pinned model that cannot satisfy the request's capabilities.
Reads the ``models`` row by ``model_id`` and fails closed on any missing
or NULL capability flag, matching the routing gate. Honors the
``cfg.routing.require_*`` config flags so a pin and routed traffic face
the same gate — see issue #2 in the capability-gate follow-ups doc.
"""
conn = _db()
try:
row = conn.execute(
"SELECT supports_vision, supports_json_mode FROM models "
"WHERE model_id = ? AND provider = 'neuralwatt'",
(model_id,),
).fetchone()
finally:
conn.close()
reason = capability_gate_reason(
dict(row) if row else {},
require_vision=cfg.routing.require_vision and caps.has_images,
require_json_mode=cfg.routing.require_json_mode and caps.require_json_mode,
)
if reason is not None:
capability = "vision" if reason.startswith("vision") else "json mode"
raise HTTPException(
422,
f"{model_id} does not support {capability}; "
f"the request {'carries image parts' if 'vision' in reason else 'requires response_format'}",
)
@app.get("/v1/models")
def list_models():
"""The routable catalog, in OpenAI's list shape.
The virtual router entries are listed first so a client that just picks
the head of the list gets routing rather than an arbitrary model.
"""
conn = _db()
try:
rows = conn.execute(
"""
SELECT model_id, latency_class FROM models
WHERE access_level IN (%s) AND availability = 'active'
ORDER BY model_id
"""
% ",".join("?" * len(cfg.routing.allowed_access_levels)),
tuple(cfg.routing.allowed_access_levels),
).fetchall()
finally:
conn.close()
data = [
{"id": ROUTER_MODEL, "object": "model", "owned_by": "router"},
{"id": ROUTER_MODEL_BATCH, "object": "model", "owned_by": "router"},
]
data += [
{"id": r["model_id"], "object": "model", "owned_by": "neuralwatt"}
for r in rows
]
return {"object": "list", "data": data}
def _sniff_telemetry_line(line: str) -> Optional[tuple[str, dict]]:
"""Parse NeuralWatt's SSE *comment* lines carrying energy/cost.
A streaming response ends with `: energy {...}` and `: cost {...}` before
`data: [DONE]`. They are SSE comments, so every ordinary client ignores
them — which is exactly why the stream can be proxied through untouched
while still being read on the way past.
"""
if not line.startswith(":"):
return None
body = line[1:].strip()
for key in ("energy", "cost"):
prefix = f"{key} "
if body.startswith(prefix):
try:
return key, json.loads(body[len(prefix):])
except json.JSONDecodeError:
return None
return None
@app.post("/v1/chat/completions")
def chat_completions(body: dict[str, Any], background: BackgroundTasks):
"""OpenAI-compatible completions, routed then proxied.
Streaming is passed through chunk by chunk rather than buffered, so a
coding agent still renders tokens as they arrive; the telemetry is
scraped from the trailing SSE comments on the way past.
"""
logs.new_trace()
started = time.perf_counter()
messages = body.get("messages") or []
if not messages:
raise HTTPException(400, "messages is required")
# Needed by every decision branch below, including the passthrough and
# local-vision paths that never reach the dispatch section, so they are
# computed up here rather than late (where they previously lived).
session_key = session_fingerprint(messages)
streamed = bool(body.get("stream"))
# What this request needs, read from the body once and used by both the
# routed and pass-through branches. Stated by the caller, not guessed at:
# a tools array, an image_url part and a response_format each say exactly
# whether a capability is required.
caps = detect_capabilities(body)
requested = body.get("model") or ROUTER_MODEL
# Some clients (opencode among them) send the model as `provider/model`,
# for a pin as much as for `auto`. Real ids can contain slashes too
# (`deepseek-ai/DeepSeek-V4-Flash`), so a stripped suffix is only trusted
# once it resolves to a real catalog row -- otherwise the id, slash and
# all, is what gets dispatched, on the chance the slash is really part of
# it. Without this, a pin sent as `llm-router/gemma-4-31b` kept its
# prefix past this point: the capability pre-check looked up a model_id
# the catalog has never heard of, found no row, and fail-closed a pin
# that could satisfy the request into a false 422 -- and even past that
# check, the same unstripped id would have gone upstream and drawn a 400
# from NeuralWatt, which only knows the bare form.
bare = requested.rsplit("/", 1)[-1]
wants_routing = bare in (ROUTER_MODEL, ROUTER_MODEL_BATCH)
if wants_routing or (bare != requested and _model_exists(bare)):
requested = bare
logs.debug(
"request",
asked=requested,
routed=wants_routing,
stream=bool(body.get("stream")),
tools=caps.tools_present,
msgs=len(messages),
capped=body.get("max_tokens"),
)
if wants_routing:
latency = BATCH if requested == ROUTER_MODEL_BATCH else INTERACTIVE
tools_present = caps.tools_present
# The turn before the last: a short follow-up inherits its complexity,
# so the classifier sees "Context: <prior>\n---\nMessage: <current>"
# instead of judging the follow-up alone.
prev_context = _previous_context(messages)
classify_started = time.perf_counter()
decision = route(
TaskRequest(
task=_last_user_text(messages),
context=prev_context,
latency_tolerance=latency,
tools_present=tools_present,
has_images=caps.has_images,
require_json_mode=caps.require_json_mode,
# Take whichever is larger: what the classifier thinks it
# needs, or what the conversation actually measures.
required_context_tokens=None,
)
)
# The classifier's own verdict, before the re-route below rewrites the
# Classification's source to 'override'.
classified_src = decision.classification.source
classifier_ms = _ms(classify_started)
ctx_src = "classifier"
# When pinch is enabled, prune ONCE before the measured-context
# decision, so the window/tier/cost choice sees the size that will
# actually ship upstream rather than the raw conversation. The pruned
# list is reused at dispatch time (send_messages below), never pruned
# twice. When pinch is off this is a byte-for-byte no-op (full list,
# as today). Pinch trims only tool results, never user/assistant/system
# messages, so the classification turn above (which used
# `_previous_context` from the full messages) is undisturbed.
if cfg.pinch.enabled:
send_messages, pinch_stats = prune_context(
list(messages),
budget_tokens=cfg.pinch.budget_tokens,
keep_last_turns=cfg.pinch.keep_last_turns,
max_summarize_chars=cfg.pinch.max_summarize_chars,
)
logs.debug(
"pinch",
pruned=pinch_stats["pruned"],
saved=pinch_stats["tokens_saved"],
orig=pinch_stats["original_tokens"],
final=pinch_stats["final_tokens"],
)
else:
send_messages = messages
measured = estimate_prompt_tokens(send_messages, tools=body.get("tools"))
if measured > decision.classification.required_context_tokens:
ctx_src = "measured"
decision = route(
TaskRequest(
task=_last_user_text(messages),
# context omitted: the override branch (task_category +
# task_tier + required_context_tokens all provided)
# never reads req.context — it skips classify().
latency_tolerance=latency,
tools_present=tools_present,
has_images=caps.has_images,
require_json_mode=caps.require_json_mode,
task_category=decision.classification.task_category,
task_tier=decision.classification.task_tier,
required_context_tokens=measured,
)
)
if decision.selected is None:
if (
caps.has_images
and not caps.require_json_mode
and cfg.local_vision.enabled
):
fallback = _run_local_vision(messages, cfg.local_vision)
if fallback is not None:
persist_route_decision(
"local_vision",
latency_tolerance=decision.latency_tolerance,
selected_model=cfg.local_vision.model,
selected_provider="local",
rejected_reason=None,
session_key=session_key,
tools=tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
streamed=streamed,
)
return _local_vision_response(
fallback, streaming=bool(body.get("stream"))
)
# Name every constraint that was applied, including the one the
# caller did not ask for explicitly. A 422 that omits the tool
# filter sends someone hunting through context and tier settings
# for a limit that came from the `tools` array.
limits = [
f"tier >= {decision.classification.task_tier}",
f"context >= {decision.classification.required_context_tokens} tokens",
decision.latency_tolerance,
]
if tools_present and cfg.routing.min_tool_proficiency is not None:
limits.append(
f"tool-use proficiency >= {cfg.routing.min_tool_proficiency} "
"(request carries tool definitions)"
)
if caps.has_images:
limits.append(
"vision-capable model (request carries image(s))"
)
if caps.require_json_mode:
limits.append(
"json-mode-capable model (response_format requires it)"
)
logs.warning("no_candidate", limits="; ".join(limits))
persist_route_decision(
"chat",
classification=decision,
rejected_reason="; ".join(limits),
session_key=session_key,
tools=tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
streamed=streamed,
)
raise HTTPException(
422,
"No model satisfies the hard filters for this request ("
+ ", ".join(limits)
+ ").",
)
log_decision(
decision,
tools=tools_present,
ms=_ms(started),
src=classified_src,
ctx_src=ctx_src,
)
persist_route_decision(
"chat",
classification=decision,
session_key=session_key,
tools=tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
streamed=streamed,
classification_source=classified_src,
classifier_ms=classifier_ms,
)
target = decision.selected.model_id
provider = decision.selected.provider
category = decision.classification.task_category
else:
target, provider, category = requested, "neuralwatt", "general_chat"
# Not a routing decision at all, and worth saying so plainly: a client
# pinned to one model gets none of the filtering or ranking below.
logs.info("passthrough", model=target, stream=bool(body.get("stream")))
persist_route_decision(
"passthrough",
selected_model=target,
selected_provider=provider,
rejected_reason=None,
session_key=session_key,
tools=int(caps.tools_present),
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
streamed=streamed,
)
# A pinned model id is dispatched as asked, but a pin that cannot
# possibly satisfy the request should fail with a clear 422 instead of
# an opaque 400 from the provider. The pin could never have worked
# anyway, so nothing is lost by rejecting early.
if caps.has_images or caps.require_json_mode:
_check_pinned_capabilities(requested, caps)
settings = cfg.dispatch_providers[provider]
api_key = os.environ.get(settings.api_key_env)
if not api_key:
raise HTTPException(503, f"{settings.api_key_env} is not set.")
# A truncated answer under a cap the CLIENT chose is not the model failing.
client_capped = body.get("max_tokens") is not None
# What session this dispatch belongs to; the hashed key is computed at the
# top (needed by the passthrough/local-vision branches), only the working
# directory is derived here because only the observation path uses it.
session_dir = session_directory(messages)
if not wants_routing:
# The routed path pruned send_messages once, before its measured-context
# decision, and the pruned list is reused here. A passthrough request
# never went through that path, so it prunes now — same gating as
# before (only when pinch is enabled).
send_messages = messages
if cfg.pinch.enabled:
# Optional relevance-based context pruning: when the conversation
# exceeds budget_tokens, trim old tool results BEFORE any paid token
# is sent upstream. User/assistant/system messages are always kept;
# only the classifier input is exempt (it already clamps to head+tail).
send_messages, pinch_stats = prune_context(
list(messages),
budget_tokens=cfg.pinch.budget_tokens,
keep_last_turns=cfg.pinch.keep_last_turns,
max_summarize_chars=cfg.pinch.max_summarize_chars,
)
logs.debug(
"pinch",
pruned=pinch_stats["pruned"],
saved=pinch_stats["tokens_saved"],
orig=pinch_stats["original_tokens"],
final=pinch_stats["final_tokens"],
)
upstream_body = {**body, "model": target, "messages": list(send_messages)}
streaming = bool(body.get("stream"))
if streaming:
# Without this the final chunk carries no usage and the observation
# would be logged with null token counts.
upstream_body.setdefault("stream_options", {"include_usage": True})
url = f"{settings.base_url}/chat/completions"
headers = {"authorization": f"Bearer {api_key}"}
if not streaming:
# A tier's iteration budget: corrective attempts AFTER a verification
# failure, never speculative ones. Retries are matched to the failure —
# truncation gets a bigger budget on the same model, malformed output
# escalates to the next candidate. See iteration.py.
budget = (
attempts_allowed(
decision.classification.task_tier,
decision.latency_tolerance,
cfg.iteration.attempts_by_tier,
cfg.iteration.max_attempts_interactive,
)
if cfg.iteration.enabled and wants_routing
else 0
)
# Bound to the routed case. A pass-through request never produced a
# `decision`, so reading its runners-up raised NameError before the
# provider was ever called — every non-streaming request naming a real
# model id returned 500. Empty is also the right value: there is no
# ranking behind a model the caller named itself, and `budget` is 0 for
# a pass-through, so the loop breaks on its first pass regardless.
alternatives = (
[
(c.model_id, model_output_ceiling(c.model_id, provider))
for c in decision.runners_up
]
if wants_routing
else []
)
current_model = target
current_max_tokens = body.get("max_tokens")
attempts_used = 0
retry_trail: list[str] = []
while True:
attempt_body = {
**body, "model": current_model,
"messages": list(send_messages),
}
if current_max_tokens is not None:
attempt_body["max_tokens"] = current_max_tokens
upstream_started = time.perf_counter()
resp = requests.post(url, headers=headers, json=attempt_body, timeout=600)
upstream_ms = _ms(upstream_started)
if resp.status_code >= 400:
logs.error(
"upstream",
model=current_model,
status=resp.status_code,
detail=resp.text[:200],
ms=upstream_ms,
)
raise HTTPException(resp.status_code, resp.text[:500])
payload = resp.json()
usage = payload.get("usage") or {}
completion_tokens = usage.get("completion_tokens")
request_id = payload.get("id")
telemetry = extract_telemetry(payload)
log_observation(
current_model, provider, category, request_id,
session_key, session_dir,
prompt_tokens=usage.get("prompt_tokens"),
completion_tokens=completion_tokens,
telemetry=telemetry,
)
choice = (payload.get("choices") or [{}])[0]
content = (choice.get("message") or {}).get("content") or ""
has_tools = bool((choice.get("message") or {}).get("tool_calls"))
result = verify_response(
content, choice.get("finish_reason"), has_tool_calls=has_tools
)
log_verification(
current_model, provider, category, request_id,
kind="structural", verdict=result.verdict, detail=result.detail,
completion_tokens=completion_tokens,
# A client's token cap explains a TRUNCATED answer, nothing
# else. Applied to every verdict it silently excluded all
# traffic from clients that always set max_tokens — opencode
# does — which made real failures invisible to feedback.
model_attributable=not (
client_capped and result.verdict == "truncated"
),
)
logs.info(
"dispatch",
model=current_model,
rid=request_id,
sess=session_key,
cat=category,
p_tok=usage.get("prompt_tokens"),
c_tok=completion_tokens,
kwh=telemetry.energy_kwh,
usd=telemetry.cost_usd,
verdict=result.verdict,
attempt=attempts_used + 1,
upstream_ms=upstream_ms,
total_ms=_ms(started),
)
if not result.failed or attempts_used >= budget:
break
plan = plan_retry(
result.verdict,
current_model,
current_max_tokens,
model_output_ceiling(current_model, provider),
alternatives,
client_capped,
)
if plan is None:
# Declining is a real outcome: a retry that cannot help still
# spends energy against a fixed quota.
break
retry_trail.append(f"{result.verdict}:{plan.detail}")
logs.warning(
"retry",
verdict=result.verdict,
reason=plan.reason,
to=plan.model_id,
detail=plan.detail,
)
if plan.reason == "malformed":
alternatives = alternatives[1:]
current_model = plan.model_id
current_max_tokens = plan.max_tokens
attempts_used += 1
if cfg.verification.local_llm_enabled and worth_local_check(
result.verdict, completion_tokens,
cfg.verification.min_completion_tokens, has_tools,
):
background.add_task(
run_local_verification, current_model, provider, category,
_last_user_text(messages), content, completion_tokens,
request_id,
)
# Report the model actually used, so the client isn't told 'auto'.
payload["model"] = current_model
# Surfaced as headers rather than in the body: the body must stay a
# valid OpenAI response, and a client that ignores headers is
# unaffected.
return JSONResponse(
content=payload,
headers={
"X-Router-Model": current_model,
"X-Router-Verification": result.verdict,
"X-Router-Attempts": str(attempts_used + 1),
**({"X-Router-Retries": "; ".join(retry_trail)} if retry_trail else {}),
},
)
# Bound OUT here, not read inside. starlette resumes a StreamingResponse's
# generator through a threadpool, and every next() runs in a fresh COPY of
# the caller's context, so a ContextVar set inside is gone by the following
# resumption -- including the finally block that logs the dispatch line.
# Measured: it logged id=- until the trace was carried explicitly. This is
# the path all agent traffic takes.
slog = logs.bind()
def proxy():
stream_started = time.perf_counter()
collected: dict[str, dict] = {}
usage: dict = {}
# Accumulated so a streamed answer gets the same structural check as a
# buffered one. Without this, streaming — which is how opencode and
# every other agent talks to the router — would be the unverified path.
content_parts: list[str] = []
finish_reason: Optional[str] = None
stream_request_id: Optional[str] = None
stream_tool_calls = False
upstream = requests.post(
url, headers=headers, json=upstream_body, stream=True, timeout=600
)
if upstream.status_code >= 400:
detail = upstream.text[:500]
slog.error(
"upstream",
model=target,
status=upstream.status_code,
detail=detail[:200],
stream=True,
ms=_ms(stream_started),
)
yield f"data: {json.dumps({'error': {'message': detail}})}\n\n".encode()
return
try:
for raw in upstream.iter_lines(decode_unicode=True):
if raw is None:
continue
sniffed = _sniff_telemetry_line(raw)
if sniffed:
collected[sniffed[0]] = sniffed[1]
elif raw.startswith("data: ") and raw.strip() != "data: [DONE]":
try:
chunk = json.loads(raw[6:])
if chunk.get("id"):
stream_request_id = chunk["id"]
if chunk.get("usage"):
usage = chunk["usage"]
for ch in chunk.get("choices") or []:
if (ch.get("delta") or {}).get("tool_calls"):
stream_tool_calls = True
piece = (ch.get("delta") or {}).get("content")
if piece:
content_parts.append(piece)
if ch.get("finish_reason"):
finish_reason = ch["finish_reason"]
except json.JSONDecodeError:
pass
yield f"{raw}\n".encode()
except requests.exceptions.RequestException as exc:
# The upstream connection can die mid-stream (NeuralWatt closing
# early, a network blip). Left uncaught, this propagates out of
# the generator and Starlette surfaces it as an unhandled ASGI
# exception -- a full traceback in the log and the client's
# connection just cut dead with no [DONE]. Whatever was collected
# before the break still gets logged below via `finally`.
slog.error(
"upstream_stream_broken",
model=target,
rid=stream_request_id,
error=str(exc),
ms=_ms(stream_started),
)
yield f"data: {json.dumps({'error': {'message': f'upstream stream interrupted: {exc}'}})}\n\n".encode()
yield b"data: [DONE]\n\n"
finally:
upstream.close()
# Logged even on a client disconnect — the energy was spent.
telemetry = extract_telemetry(collected)
log_observation(
target, provider, category, stream_request_id,
session_key, session_dir,
prompt_tokens=usage.get("prompt_tokens"),
completion_tokens=usage.get("completion_tokens"),
telemetry=telemetry,
)
streamed = "".join(content_parts)
result = verify_response(
streamed, finish_reason, has_tool_calls=stream_tool_calls
)
ct = usage.get("completion_tokens")
slog.info(
"dispatch",
model=target,
rid=stream_request_id,
sess=session_key,
cat=category,
stream=True,
p_tok=usage.get("prompt_tokens"),
c_tok=ct,
kwh=telemetry.energy_kwh,
usd=telemetry.cost_usd,
verdict=result.verdict,
total_ms=_ms(stream_started),
)
log_verification(
target, provider, category, stream_request_id,
kind="structural", verdict=result.verdict, detail=result.detail,
completion_tokens=ct,
# A client's token cap explains a TRUNCATED answer, nothing
# else. Applied to every verdict it silently excluded all
# traffic from clients that always set max_tokens — opencode
# does — which made real failures invisible to feedback.
model_attributable=not (
client_capped and result.verdict == "truncated"
),
)
# Inline rather than a background task: the generator has already
# finished streaming, so the client is not waiting on this.
if cfg.verification.local_llm_enabled and worth_local_check(
result.verdict, ct, cfg.verification.min_completion_tokens,
stream_tool_calls,
):
run_local_verification(
target, provider, category,
_last_user_text(messages), streamed, ct,
stream_request_id,
)
return StreamingResponse(proxy(), media_type="text/event-stream")
@app.post("/dispatch", response_model=DispatchResponse)
def dispatch_endpoint(req: TaskRequest):
logs.new_trace()
started = time.perf_counter()
decision = route(req)
log_decision(
decision,
tools=req.tools_present,
ms=_ms(started),
ctx_src="caller" if req.required_context_tokens is not None else "classifier",
)
persist_route_decision(
"dispatch",
classification=decision,
tools=req.tools_present,
images=req.has_images,
json_mode=req.require_json_mode,
classifier_ms=_ms(started),
)
if decision.selected is None:
raise HTTPException(
422,
"No model satisfies the hard filters for this task "
f"(tier >= {decision.classification.task_tier}, context >= "
f"{decision.classification.required_context_tokens} tokens, "
f"{decision.latency_tolerance}). Trim the context or widen "
"routing.allowed_access_levels.",
)
selected = decision.selected
client = _provider_client(selected.provider)
messages = [{"role": "user", "content": req.task}]
if req.context:
messages.insert(0, {"role": "system", "content": req.context})
upstream_started = time.perf_counter()
try:
# with_raw_response because the energy and cost blocks sit outside the
# OpenAI schema and the parsed object drops them.
raw = client.chat.completions.with_raw_response.create(
model=selected.model_id, messages=messages
)
except OpenAIError as e:
logs.error(
"upstream",
model=selected.model_id,
error=type(e).__name__,
detail=str(e)[:200],
ms=_ms(upstream_started),
)
raise HTTPException(502, f"Dispatch to {selected.model_id} failed: {e}")
payload = json.loads(raw.text)
usage = payload.get("usage") or {}
prompt_tokens = usage.get("prompt_tokens")
completion_tokens = usage.get("completion_tokens")
telemetry = extract_telemetry(payload)
log_observation(
selected.model_id,
selected.provider,
decision.classification.task_category,
payload.get("id"),
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
telemetry=telemetry,
)
choices = payload.get("choices") or [{}]
content = (choices[0].get("message") or {}).get("content") or ""
logs.info(
"dispatch",
model=selected.model_id,
rid=payload.get("id"),
p_tok=prompt_tokens,
c_tok=completion_tokens,
kwh=telemetry.energy_kwh,
usd=telemetry.cost_usd,
upstream_ms=_ms(upstream_started),
total_ms=_ms(started),
)
return DispatchResponse(
route=decision,
content=content,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
telemetry=telemetry,
)
Reference in New Issue View Git Blame Copy Permalink