5f7716e121
Address the 11 confirmed bugs and 4 cleanups from context-pruning-and-framing-review.md, plus the image-accounting regression it found on a second pass. classifier framing: - _previous_context walks back to the nearest assistant turn only, skipping system/user/tool so raw tool output and the system prompt never contaminate the framing signal; reuses context_prune.extract_text instead of a duplicated block parser. - classify() drops its redundant if/not-context branch. - framing instruction moved out of the static system prompt and into _classifier_user_content so it travels only with content that justifies it (respects context_framing opt-out). context pruning: - extract_text counts image_url bytes so image-bearing tool results are sized (and pruned) correctly. - -- new, follow-up review -- image_url blocks now count toward the size trigger but not toward the replacement text, and are stubbed to a short marker when the message is trimmed, so an image-heavy tool result genuinely shrinks rather than staying full-size while tokens_saved lied and base64 leaked into the text block. - tokens_saved is now the honest whole-list before/after reduction (max(orig - final, 0)), never a per-message estimate that could drift. - max_summarize_chars validated >= 3000 at config load; trim branch guards against negative math and message growth. - recency guard protects the newest tool result even with no user turn. - non-text blocks survive trimming (type preserved) rather than being flattened away. - dropped/stat terminology corrected to summarized; _turn_of() removed; module-level pinch defaults removed in favor of PinchConfig. Dispatch and config: - prune_context runs once before the measured-context routing decision (reused at dispatch, no double-prune, no in-place mutation); no-op when pinch is disabled. - dead context=prev_context arg removed from the measured reroute. - TaskRequest.context dual use documented. 15 new regression tests pin the fixes; 545 tests pass.
2521 lines
100 KiB
Python
2521 lines
100 KiB
Python
#!/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 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
|
|
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)
|
|
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
datetime.now(timezone.utc).isoformat(),
|
|
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)),
|
|
),
|
|
)
|
|
conn.commit()
|
|
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()
|
|
|
|
|
|
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,
|
|
)
|