alee/6krrt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

conversation-identity: the router learns which conversation each request belongs to #99

Merged
alee merged 18 commits from feat/conversation-identity into main 2026-09-22 22:39:07 +00:00
Conversation 0 Commits 18 Files Changed 26 +2717 -154
26 changed files with 2717 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
config/config.yaml
View File

@@ -237,6 +237,13 @@ objective:
# concrete float when the penalty is engaged.
incumbent_rate_min_observations: 25
# Window (seconds) for the conversation adoption counter in /metrics.
# Without this, the counter may report stale adoption figures from a period
# when the feature was still rolling out. Null or absent = no windowing
# (all time); 0 is rejected by the config validator. For a rolling 7-day
# window: 604800.
adoption_window_seconds: 604800
# Credit-aware routing attenuation (OFF BY DEFAULT).
# When enabled, a provider configured with a balance_url (e.g. OpenRouter's
# prepaid account balance) gets its *comparison cost* inflated inside the

27
config/schema.sql
View File

@@ -266,9 +266,10 @@ CREATE TABLE IF NOT EXISTS verifications (
-- filter shut it out.
--
-- This is an OBSERVABILITY table, not a scoring input: nothing in routing.py
-- reads it. It stores only the hashed session fingerprint in `session_key`;
-- never `session_dir` and never any prompt or answer text. A test enforces
-- that the write path stores only the hash.
-- reads it. It stores the hashed session fingerprint in `session_key`, or
-- `"c:" + conversation id` when the client sent X-Router-Conversation; never
-- `session_dir` and never any prompt or answer text. A test enforces that
-- the headerless write path stores the 16-char hash.
--
-- Like `energy_observations`, `observed_at` uses
-- datetime.now(timezone.utc).isoformat().
@@ -303,7 +304,9 @@ CREATE TABLE IF NOT EXISTS route_decisions (
est_cost_usd REAL,
est_proficiency REAL,
rejected_reason TEXT, -- the 422 limits when no selection
session_key TEXT, -- hashed session fingerprint ONLY,
session_key TEXT, -- hashed session fingerprint, or
-- 'c:' + conversation id when the
-- client sent X-Router-Conversation;
-- never session_dir or prompt text
tools INTEGER, -- 0/1
images INTEGER, -- 0/1
@@ -333,10 +336,19 @@ CREATE TABLE IF NOT EXISTS route_decisions (
-- = the longest stable prefix
prefix_tokens_after_divergence INTEGER, -- estimated tokens at/after it
-- in THIS turn: the re-billed part
prefix_prev_message_count INTEGER -- last turn's message count. A
prefix_prev_message_count INTEGER, -- last turn's message count. A
-- pure append diverges at exactly
-- this index; anything lower is
-- rewritten history.
agent TEXT, -- client agent name from
-- X-Router-Agent (slugged by the
-- plugin, e.g. atlas-plan-executor);
-- NULL when the client sends none
parent_key TEXT -- 'c:' plus the parent conversation
-- id from X-Router-Parent, set on a
-- sub-agent's rows; NULL for
-- top-level conversations or when
-- unknown
);
CREATE INDEX IF NOT EXISTS idx_verifications_model ON verifications (model_id, provider);
@@ -381,6 +393,11 @@ CREATE TABLE IF NOT EXISTS local_energy_observations (
-- Working directory of the conversation, when derivable from messages.
-- Same purpose as energy_observations.session_dir.
session_dir TEXT,
-- 'c:<conversation id>' when the client sent X-Router-Conversation, else
-- the content fingerprint, same as route_decisions.session_key -- lets
-- local-dispatch answers be attributed to the same conversation as
-- cloud-routed ones.
session_key TEXT,
avg_power_watts REAL,
duration_seconds REAL,
energy_kwh REAL,

46
deploy/README.md
View File

@@ -236,3 +236,49 @@ routable catalog. The router hard-filters on the measured conversation size,
so a prompt too big for the smaller models simply won't be routed to them;
if it fits nothing, `/v1/chat/completions` returns a 422 naming the
constraint rather than truncating.
## opencode plugin
[`deploy/opencode-plugin/router-link.js`](opencode-plugin/router-link.js) is
an opencode plugin that does two things:
1. **Stamps conversation identity** on outbound requests to the router:
`X-Router-Conversation` (the session id), `X-Router-Agent` (the agent
name), and `X-Router-Parent` (the session's parent, if any). Only the
`llm-router` provider gets these headers — another provider (Anthropic,
OpenAI, …) would not understand them and they would just leak identity.
Agent names are slugged to the router's header charset (lowercase; runs of
characters outside `a-z0-9._:-` become `-`, trimmed, capped at 64), so a
multi-word name like `Sisyphus - ultraworker` arrives as
`sisyphus-ultraworker` instead of being dropped as invalid.
2. **Reports test outcomes** back to the router's `/outcome` endpoint so
`feedback.py` can fold client-side pass/fail into proficiency scoring.
(`tool.execute.after` hook — watches for `pytest`, `cargo test`, `go test`,
`tsc`, `ruff`, `eslint`, etc.)
### Install
```bash
# cp is aliased to -i in the default zsh; `command cp` bypasses the alias.
command cp -f deploy/opencode-plugin/router-link.js ~/.config/opencode/plugins/
rm ~/.config/opencode/plugins/router-outcome.js # <-- MUST remove this
```
**The old file must be removed** because both plugins hook
`tool.execute.after`. Keeping both would post every outcome twice, and
`feedback.py` would count each test result twice — once for the real result
and once for a duplicate that looks identical but belongs to a different
model in a different conversation. A test counted twice penalises the
incumbent model (it accrues extra failures) while never appearing in the
challenger's ledger, which is the worst attribution error in the system.
### Turn it off
Remove the file from `~/.config/opencode/plugins/` (or the per-project
`.opencode/plugins/`). Without a plugin, the router falls back to the
fingerprint-based heuristics it used before: the working directory from the
`/v1/chat/completions` request body's `source` field (set by the opencode
SDK), matched against the `session.directory` in the `/outcome` body. That
is less accurate (it cannot distinguish two sessions in the same directory),
but it is functional.

221
deploy/opencode-plugin/router-link.js Normal file
View File

@@ -0,0 +1,221 @@
/**
* Stamps requests to the local LLM router with conversation identity, and
* reports test outcomes back to it.
*
* This replaces `router-outcome.js`, which only reported outcomes. A router
* that knows *which conversation* and *which agent* a request belongs to can
* attribute quality per-session and distinguish agent-specific behavior from
* general model quality. That is the loop this plugin closes on the way in.
*
* Two hooks:
*
* chat.headers — for requests routed to the llm-router provider, set
* X-Router-Conversation (the session id),
* X-Router-Agent (the agent, slugged to the
* router's charset), and X-Router-Parent
* (the conversation's parent session, when known).
* Only the llm-router provider gets these: another
* provider would not understand them and they would
* just leak identity.
*
* tool.execute.after — the old router-outcome logic, carried over
* unchanged: watch test/build commands, POST
* pass/fail to the router's /outcome endpoint so
* feedback.py folds it into proficiency. It now also
* carries conversation_id so the router can
* attribute the outcome to the right session.
*
* Failure policy: no hook may throw or stall a request more than the 250ms
* parent lookup. The router being down must never break the session — a
* reporting/stamping failure is not the user's problem.
*
* Install:
* command cp -f deploy/opencode-plugin/router-link.js ~/.config/opencode/plugins/
* rm ~/.config/opencode/plugins/router-outcome.js
*
* (`command cp` because `cp` is aliased to -i in zsh and silently does
* nothing on an existing file; and the old file MUST go — both hook
* tool.execute.after, so an outcome would be posted twice and counted twice
* by feedback.py.)
*
* Or per-project, in .opencode/plugins/.
*/
import { setTimeout as sleep } from "node:timers/promises";
const ROUTER = process.env.LLM_ROUTER_URL || "http://127.0.0.1:8080";
const PROVIDER_ID = "llm-router";
// The deterministic path that the `chat.headers` hook must stay well under.
const PARENT_TIMEOUT_MS = 250;
const PARENT_CACHE_MAX = 1000;
const TIMEOUT_SENTINEL = Symbol("parentOf timeout");
// Commands whose exit status is a real verdict on the work. Deliberately
// narrow: a failing `ls` says nothing about model quality, and a false signal
// is worse than no signal — it trains the router on noise.
const TEST_COMMAND = new RegExp(
[
"\\bpytest\\b",
"\\bunittest\\b",
"\\bnpm\\s+(run\\s+)?test\\b",
"\\bpnpm\\s+(run\\s+)?test\\b",
"\\byarn\\s+test\\b",
"\\bvitest\\b",
"\\bjest\\b",
"\\bcargo\\s+(test|check|build)\\b",
"\\bgo\\s+(test|build|vet)\\b",
"\\bmake\\s+(test|check)\\b",
"\\bmvn\\s+test\\b",
"\\bgradle\\s+test\\b",
"\\btsc\\b",
"\\bruff\\b",
"\\bmypy\\b",
"\\beslint\\b",
].join("|"),
);
// Failure signatures, for tools that exit 0 while reporting failures.
const FAILURE_TEXT =
/\b(\d+\s+failed|FAILED|FAIL\b|Traceback \(most recent call last\)|error(s)?:|panic:|AssertionError|✗|✖)/;
function looksFailed(output) {
const exit = output?.exitCode ?? output?.exit_code;
if (typeof exit === "number" && exit !== 0) return true;
const text = `${output?.stdout ?? ""}\n${output?.stderr ?? ""}\n${
typeof output?.output === "string" ? output.output : ""
}`;
// "0 failed" and "no errors" must not trip the failure regex.
if (/\b0 failed\b|\bno errors?\b/i.test(text)) return false;
return FAILURE_TEXT.test(text);
}
function commandOf(input) {
const args = input?.args ?? input?.arguments ?? {};
return args.command ?? args.cmd ?? args.script ?? "";
}
// Cache of successful parent lookups: session id -> parentID (or null for root
// sessions when the SDK returns no parentID). We cache successes so repeated
// requests in the same session do not keep hitting the SDK. We cache nothing on
// timeout or throw -- a transient miss should not be sticky.
// Eviction: when the Map reaches PARENT_CACHE_MAX, the oldest-inserted entry
// is evicted first (Map preserves insertion order, so the first key is oldest).
const parentCache = new Map();
function cacheSet(sessionID, parentID) {
if (!parentCache.has(sessionID) && parentCache.size >= PARENT_CACHE_MAX) {
const firstKey = parentCache.keys().next().value;
parentCache.delete(firstKey);
}
parentCache.set(sessionID, parentID);
}
// Resolve a session's parent id, racing the SDK call against a timeout so a
// slow session store can never stall request stamping. Returns undefined on
// timeout or catch; never throws. Returns null for root sessions (no parent).
async function parentOf(client, sessionID) {
if (parentCache.has(sessionID)) return parentCache.get(sessionID);
try {
const result = await Promise.race([
client.session.get({ path: { id: sessionID } }),
sleep(PARENT_TIMEOUT_MS).then(() => TIMEOUT_SENTINEL),
]);
if (result === TIMEOUT_SENTINEL) return undefined;
const parentID = result?.data?.parentID ?? null;
cacheSet(sessionID, parentID);
return parentID;
} catch {
return undefined;
}
}
// Whether a request is bound for the local router. opencode identifies the
// provider both on the model (`model.providerID`) and on the provider context
// (`provider.info.id`); if neither is populated at runtime the provider's
// base URL is a usable fallback, compared by ORIGIN: opencode.json sets
// options.baseURL to something like http://127.0.0.1:8080/v1, which never
// string-equals the bare router URL. A malformed URL on either side (the
// provider's, or a bad LLM_ROUTER_URL) fails safe: no headers.
function isRouterRequest(input, providerUrl) {
if (input?.model?.providerID === PROVIDER_ID) return true;
if (input?.provider?.info?.id === PROVIDER_ID) return true;
if (!providerUrl) return false;
try {
return new URL(providerUrl).origin === new URL(ROUTER).origin;
} catch {
return false;
}
}
// The router validates X-Router-Agent against ^[A-Za-z0-9._:-]{1,64}$ and
// treats a failing value as absent, so a live name like "Sisyphus -
// ultraworker" would silently lose attribution. Slug to the legal charset:
// lowercase; runs of characters outside [a-z0-9._:-] become one '-'; repeated
// '-' collapse; leading/trailing '-' trimmed; cut to 64. A name already in
// the charset (e.g. "oh-my-claudecode:executor") passes through unchanged.
function slugAgent(name) {
return String(name)
.toLowerCase()
.replace(/[^a-z0-9._:-]+/g, "-")
.replace(/-+/g, "-")
.replace(/^-+|-+$/g, "")
.slice(0, 64);
}
export const RouterLink = async ({ client, directory }) => {
return {
"chat.headers": async (input, output) => {
const providerUrl =
input?.provider?.info?.options?.baseURL ??
input?.provider?.info?.options?.url ??
input?.provider?.info?.api?.url;
if (!isRouterRequest(input, providerUrl)) return;
output.headers["X-Router-Conversation"] = input.sessionID;
const agent = slugAgent(input.agent ?? "");
if (agent) output.headers["X-Router-Agent"] = agent;
const parentID = await parentOf(client, input.sessionID);
if (parentID) output.headers["X-Router-Parent"] = parentID;
},
"tool.execute.after": async (input, output) => {
// Only shell-ish tools carry a command whose exit status is a verdict.
const command = commandOf(input);
if (!command || !TEST_COMMAND.test(command)) return;
const ok = !looksFailed(output);
const detail = `${command.slice(0, 120)}${ok ? " — passed" : " — failed"}`;
try {
const res = await fetch(`${ROUTER}/outcome`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
ok,
detail,
source: directory,
conversation_id: input.sessionID,
}),
// The router being down must never break the user's session.
signal: AbortSignal.timeout(3000),
});
if (res.status === 409) {
// Several sessions active and the directory did not match one the
// router had seen. Dropping the sample is the correct outcome.
console.error(
"[router-link] ambiguous session; outcome not recorded",
);
} else if (!res.ok && res.status !== 404) {
console.error(`[router-link] ${res.status} reporting outcome`);
}
} catch (err) {
// Swallowed on purpose: a reporting failure is not the user's problem.
console.error(`[router-link] could not reach ${ROUTER}: ${err.message}`);
}
},
};
};
export { parentCache };

701
deploy/opencode-plugin/router-link.test.mjs Normal file
View File

@@ -0,0 +1,701 @@
/**
* Tests for router-link.js
*
* Uses node:test (Node >= 18) with mock.global to stub fetch and the SDK
* client. Runs against a default-hook instance of RouterLink.
*
* Tests are isolated by using distinct session ids for each parent-cache
* scenario, and resetting the global fetch mock between tool.execute.after
* tests.
*/
import { describe, it, mock, before, after, beforeEach } from "node:test";
import assert from "node:assert/strict";
import { RouterLink, parentCache } from "./router-link.js";
const SESSION_ID = "ses_test123";
const AGENT = "build";
const DIRECTORY = "/home/test/project";
// Stub SDK client. `session.get` behaviour is swapped per test.
function stubClient(sessionGetImpl) {
return {
session: {
get: sessionGetImpl,
},
};
}
// Instant successful client returning ses_parent456 for any session.
function defaultClient() {
return stubClient(async () => ({
data: { parentID: "ses_parent456" },
}));
}
// ------ helpers ------
function routerHeadersInput(overrides = {}) {
return {
sessionID: SESSION_ID,
agent: AGENT,
model: { providerID: "llm-router" },
provider: { info: { id: "llm-router" } },
...overrides,
};
}
function testToolInput(overrides = {}) {
return {
tool: "bash",
sessionID: SESSION_ID,
callID: "call_001",
args: { command: "pytest tests/" },
...overrides,
};
}
function passingOutput(overrides = {}) {
return {
exitCode: 0,
stdout: "3 passed, 0 failed",
stderr: "",
...overrides,
};
}
// ------ chat.headers tests ------
describe("chat.headers", () => {
it("sets X-Router-Conversation, X-Router-Agent, X-Router-Parent for router provider", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
// Use a unique session so cache from a previous test cannot interfere
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_router_001" }),
output,
);
assert.equal(output.headers["X-Router-Conversation"], "ses_router_001");
assert.equal(output.headers["X-Router-Agent"], AGENT);
assert.equal(output.headers["X-Router-Parent"], "ses_parent456");
});
it("sets no headers for non-router provider (model + provider both override)", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({
sessionID: "ses_nonrouter_001",
model: { providerID: "openai" },
provider: { info: { id: "openai" } },
}),
output,
);
assert.deepEqual(output.headers, {});
});
it("sets no headers when provider.info.id differs even if model.providerID is absent", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_nonrouter_002",
agent: AGENT,
model: {},
provider: { info: { id: "anthropic" } },
},
output,
);
assert.deepEqual(output.headers, {});
});
it("sets no headers when neither model.providerID nor provider.info.id nor URL matches", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_nonrouter_003",
agent: AGENT,
model: {},
provider: { info: { options: { url: "http://other:9000" } } },
},
output,
);
assert.deepEqual(output.headers, {});
});
it("falls back to provider url matching when model.providerID is absent", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_urlmatch_001",
agent: AGENT,
model: {},
provider: {
info: {
options: { url: "http://127.0.0.1:8080" },
},
},
},
output,
);
assert.equal(output.headers["X-Router-Conversation"], "ses_urlmatch_001");
});
it("falls back to provider baseURL origin matching when model.providerID is absent", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_baseurl_001",
agent: AGENT,
model: {},
provider: {
info: {
options: { baseURL: "http://127.0.0.1:8080/v1" },
},
},
},
output,
);
assert.equal(output.headers["X-Router-Conversation"], "ses_baseurl_001");
});
it("sets no headers when the baseURL origin differs from the router origin", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_baseurl_002",
agent: AGENT,
model: {},
provider: {
info: {
options: { baseURL: "http://127.0.0.1:9999/v1" },
},
},
},
output,
);
assert.deepEqual(output.headers, {});
});
it("sets no headers on a malformed baseURL and does not throw", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_baseurl_003",
agent: AGENT,
model: {},
provider: {
info: {
options: { baseURL: "not-a-valid-url" },
},
},
},
output,
);
assert.deepEqual(output.headers, {});
});
it("omits X-Router-Parent when parent lookup takes >250ms", async () => {
const slowClient = stubClient(
() =>
new Promise((resolve) =>
setTimeout(() => resolve({ data: { parentID: "ses_slow" } }), 300),
),
);
const hooks = await RouterLink({
client: slowClient,
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_slowparent_001" }),
output,
);
assert.equal(output.headers["X-Router-Conversation"], "ses_slowparent_001");
assert.equal(output.headers["X-Router-Agent"], AGENT);
// Parent timed out, so no parent header
assert.equal(output.headers["X-Router-Parent"], undefined);
});
it("omits X-Router-Parent when parent lookup throws and does not throw", async () => {
const brokenClient = stubClient(async () => {
throw new Error("session store unreachable");
});
const hooks = await RouterLink({
client: brokenClient,
directory: DIRECTORY,
});
const output = { headers: {} };
// Must not throw
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_throws_001" }),
output,
);
assert.equal(output.headers["X-Router-Conversation"], "ses_throws_001");
assert.equal(output.headers["X-Router-Agent"], AGENT);
assert.equal(output.headers["X-Router-Parent"], undefined);
});
it("caches successful parent lookups so repeated calls do not hit the SDK", async () => {
let callCount = 0;
const countingClient = stubClient(async () => {
callCount++;
return { data: { parentID: "ses_parentCached" } };
});
const hooks = await RouterLink({
client: countingClient,
directory: DIRECTORY,
});
const output1 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_counting_001" }),
output1,
);
assert.equal(callCount, 1);
// Second call with same session id should use cache
const output2 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_counting_001" }),
output2,
);
assert.equal(callCount, 1); // still 1 — cache hit
assert.equal(output2.headers["X-Router-Parent"], "ses_parentCached");
});
it("does not cache failed lookups (so a retry can try again)", async () => {
let fail = true;
const flakyClient = stubClient(async () => {
if (fail) throw new Error("transient");
return { data: { parentID: "ses_parentFlaky" } };
});
const hooks = await RouterLink({
client: flakyClient,
directory: DIRECTORY,
});
// First call: fails, no cache
const output1 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_flaky_001" }),
output1,
);
assert.equal(output1.headers["X-Router-Parent"], undefined);
// Second call: SDK succeeds — should NOT have cached the failure
fail = false;
const output2 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_flaky_001" }),
output2,
);
assert.equal(output2.headers["X-Router-Parent"], "ses_parentFlaky");
});
});
// ------ chat.headers agent slugging ------
describe("chat.headers agent slugging", () => {
it("slugs 'Atlas - Plan Executor' to 'atlas-plan-executor'", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({
sessionID: "ses_slug_001",
agent: "Atlas - Plan Executor",
}),
output,
);
assert.equal(output.headers["X-Router-Agent"], "atlas-plan-executor");
});
it("slugs 'Sisyphus - ultraworker' to 'sisyphus-ultraworker'", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({
sessionID: "ses_slug_002",
agent: "Sisyphus - ultraworker",
}),
output,
);
assert.equal(output.headers["X-Router-Agent"], "sisyphus-ultraworker");
});
it("leaves a name already in the router charset unchanged", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({
sessionID: "ses_slug_003",
agent: "oh-my-claudecode:executor",
}),
output,
);
assert.equal(
output.headers["X-Router-Agent"],
"oh-my-claudecode:executor",
);
});
it("sends no X-Router-Agent when the name is only spaces", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_slug_004", agent: " " }),
output,
);
// The hook itself still ran and stamped the conversation.
assert.equal(output.headers["X-Router-Conversation"], "ses_slug_004");
assert.equal(output.headers["X-Router-Agent"], undefined);
});
it("cuts a 100-character name to the router's 64-char limit", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({
sessionID: "ses_slug_005",
agent: "a".repeat(100),
}),
output,
);
assert.equal(output.headers["X-Router-Agent"], "a".repeat(64));
});
it("sends no X-Router-Agent when input.agent is absent", async () => {
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const output = { headers: {} };
await hooks["chat.headers"](
{
sessionID: "ses_slug_006",
model: { providerID: "llm-router" },
provider: { info: { id: "llm-router" } },
},
output,
);
assert.equal(output.headers["X-Router-Agent"], undefined);
});
});
// ------ tool.execute.after tests ------
describe("tool.execute.after", () => {
beforeEach(() => {
mock.restoreAll();
});
it("POSTs pass outcome with conversation_id, ok=true, and source", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
fetchMock.mock.mockImplementation(async () =>
new Response(null, { status: 200 }),
);
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
await hooks["tool.execute.after"](
testToolInput({ sessionID: "ses_outcome_001" }),
passingOutput(),
);
assert.equal(fetchMock.mock.callCount(), 1);
const [url, opts] = fetchMock.mock.calls[0].arguments;
assert.match(url, /\/outcome$/);
const body = JSON.parse(opts.body);
assert.equal(body.ok, true);
assert.equal(body.source, DIRECTORY);
assert.equal(body.conversation_id, "ses_outcome_001");
assert.match(body.detail, /passed/);
});
it("POSTs fail outcome with ok=false when tool exits non-zero", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
fetchMock.mock.mockImplementation(async () =>
new Response(null, { status: 200 }),
);
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
const input = testToolInput({ sessionID: "ses_outcome_002" });
const output = passingOutput({ exitCode: 1, stdout: "1 failed" });
await hooks["tool.execute.after"](input, output);
assert.equal(fetchMock.mock.callCount(), 1);
const body = JSON.parse(fetchMock.mock.calls[0].arguments[1].body);
assert.equal(body.ok, false);
assert.match(body.detail, /failed/);
});
it("POSTs nothing for a non-test command", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
await hooks["tool.execute.after"](
testToolInput({
sessionID: "ses_other_001",
args: { command: "ls -la" },
}),
passingOutput(),
);
assert.equal(fetchMock.mock.callCount(), 0);
});
it("does not throw when router is unreachable (network error)", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
fetchMock.mock.mockImplementation(async () => {
throw new TypeError("fetch failed");
});
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
// Must not throw
await hooks["tool.execute.after"](
testToolInput({ sessionID: "ses_unreach_001" }),
passingOutput(),
);
assert.equal(fetchMock.mock.callCount(), 1);
});
it("logs and swallows a 409 response", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
fetchMock.mock.mockImplementation(async () =>
new Response(null, { status: 409 }),
);
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
// Must not throw
await hooks["tool.execute.after"](
testToolInput({ sessionID: "ses_409_001" }),
passingOutput(),
);
assert.equal(fetchMock.mock.callCount(), 1);
});
it("logs non-ok 4xx (but not 404) without throwing", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
fetchMock.mock.mockImplementation(async () =>
new Response(null, { status: 502 }),
);
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
// Must not throw
await hooks["tool.execute.after"](
testToolInput({ sessionID: "ses_502_001" }),
passingOutput(),
);
assert.equal(fetchMock.mock.callCount(), 1);
});
it("accepts 404 silently (old router without /outcome)", async () => {
const fetchMock = mock.fn();
mock.method(globalThis, "fetch", fetchMock);
fetchMock.mock.mockImplementation(async () =>
new Response(null, { status: 404 }),
);
const hooks = await RouterLink({
client: defaultClient(),
directory: DIRECTORY,
});
await hooks["tool.execute.after"](
testToolInput({ sessionID: "ses_404_001" }),
passingOutput(),
);
assert.equal(fetchMock.mock.callCount(), 1);
});
});
// ------ parent cache tests ------
describe("parent cache", () => {
it("caches a root session (no parentID) so two requests call the SDK once", async () => {
let callCount = 0;
const countingClient = stubClient(async () => {
callCount++;
return { data: {} }; // no parentID -- root session
});
const hooks = await RouterLink({
client: countingClient,
directory: DIRECTORY,
});
// First call: SDK hit, null cached, no X-Router-Parent
const output1 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_root_001" }),
output1,
);
assert.equal(callCount, 1);
assert.equal(output1.headers["X-Router-Parent"], undefined);
// Second call with same session: cache hit, SDK NOT called
const output2 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_root_001" }),
output2,
);
assert.equal(callCount, 1); // still 1
assert.equal(output2.headers["X-Router-Parent"], undefined);
});
it("retries a timed-out lookup on the next request", async () => {
let callCount = 0;
const slowClient = stubClient(() => {
callCount++; // count synchronously on each call
return new Promise((resolve) =>
setTimeout(() => resolve({ data: { parentID: "ses_parentSlow" } }), 300),
);
});
const hooks = await RouterLink({
client: slowClient,
directory: DIRECTORY,
});
// First call: timeout wins (300 ms > 250 ms), SDK call counted
const output1 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_timoutretry_001" }),
output1,
);
assert.equal(callCount, 1);
assert.equal(output1.headers["X-Router-Parent"], undefined);
// Second call: not cached, SDK called again
const output2 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_timoutretry_001" }),
output2,
);
assert.equal(callCount, 2); // SDK retried
assert.equal(output2.headers["X-Router-Parent"], undefined);
});
it("retries a throwing lookup on the next request", async () => {
let callCount = 0;
const brokenClient = stubClient(() => {
callCount++; // count synchronously before rejecting
return Promise.reject(new Error("session store down"));
});
const hooks = await RouterLink({
client: brokenClient,
directory: DIRECTORY,
});
// First call: throws, no cache
const output1 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_throwretry_001" }),
output1,
);
assert.equal(callCount, 1);
assert.equal(output1.headers["X-Router-Parent"], undefined);
// Second call: not cached, SDK called again
const output2 = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: "ses_throwretry_001" }),
output2,
);
assert.equal(callCount, 2); // SDK retried
assert.equal(output2.headers["X-Router-Parent"], undefined);
});
it("evicts the oldest entry when cache exceeds 1000", async () => {
parentCache.clear();
const noParentClient = stubClient(async () => ({ data: {} }));
const hooks = await RouterLink({
client: noParentClient,
directory: DIRECTORY,
});
// Insert 1001 entries
for (let i = 0; i < 1001; i++) {
const output = { headers: {} };
await hooks["chat.headers"](
routerHeadersInput({ sessionID: `ses_evict_${String(i).padStart(4, "0")}` }),
output,
);
}
assert.equal(parentCache.size, 1000);
// Oldest entry (ses_evict_0000) evicted
assert.equal(parentCache.has("ses_evict_0000"), false);
// Newest entry still present
assert.equal(parentCache.has("ses_evict_1000"), true);
// An entry in the middle still present
assert.equal(parentCache.has("ses_evict_0500"), true);
});
});

108
deploy/opencode-plugin/router-outcome.js
View File

@@ -1,108 +0,0 @@
/**
* Report test outcomes back to the local LLM router.
*
* This closes the only loop the router cannot close by itself. Everything it
* records on its own is a proxy: structural checks know whether code *parses*,
* the local checker guesses whether prose *looks* right. Neither knows whether
* the answer did the job. opencode does — it runs your tests.
*
* Hooks `tool.execute.after`, watches for test/build commands, and POSTs the
* pass/fail to the router's /outcome endpoint. The router folds client
* outcomes into proficiency in BOTH directions (a client's "succeeded" means
* the work worked, unlike a parser's "ok" which only means it parsed), so this
* is what eventually makes routing discriminate on quality.
*
* Install:
* mkdir -p ~/.config/opencode/plugins
* cp router-outcome.js ~/.config/opencode/plugins/
*
* Or per-project, in .opencode/plugins/.
*
* PARALLEL SESSIONS: the report carries this session's directory, which the
* router matches against the working directory it sees in the conversation.
* That is exact even with several sessions running. If it cannot match, and
* more than one conversation has been routed recently, the router answers 409
* and records nothing rather than guessing — a misattributed failure would
* penalize a model for work it never did.
*/
const ROUTER = process.env.LLM_ROUTER_URL || "http://127.0.0.1:8080";
// Commands whose exit status is a real verdict on the work. Deliberately
// narrow: a failing `ls` says nothing about model quality, and a false signal
// is worse than no signal — it trains the router on noise.
const TEST_COMMAND = new RegExp(
[
"\\bpytest\\b",
"\\bunittest\\b",
"\\bnpm\\s+(run\\s+)?test\\b",
"\\bpnpm\\s+(run\\s+)?test\\b",
"\\byarn\\s+test\\b",
"\\bvitest\\b",
"\\bjest\\b",
"\\bcargo\\s+(test|check|build)\\b",
"\\bgo\\s+(test|build|vet)\\b",
"\\bmake\\s+(test|check)\\b",
"\\bmvn\\s+test\\b",
"\\bgradle\\s+test\\b",
"\\btsc\\b",
"\\bruff\\b",
"\\bmypy\\b",
"\\beslint\\b",
].join("|"),
);
// Failure signatures, for tools that exit 0 while reporting failures.
const FAILURE_TEXT =
/\b(\d+\s+failed|FAILED|FAIL\b|Traceback \(most recent call last\)|error(s)?:|panic:|AssertionError|✗|✖)/;
function looksFailed(output) {
const exit = output?.exitCode ?? output?.exit_code;
if (typeof exit === "number" && exit !== 0) return true;
const text = `${output?.stdout ?? ""}\n${output?.stderr ?? ""}\n${
typeof output?.output === "string" ? output.output : ""
}`;
// "0 failed" and "no errors" must not trip the failure regex.
if (/\b0 failed\b|\bno errors?\b/i.test(text)) return false;
return FAILURE_TEXT.test(text);
}
function commandOf(input) {
const args = input?.args ?? input?.arguments ?? {};
return args.command ?? args.cmd ?? args.script ?? "";
}
export const RouterOutcome = async ({ directory }) => {
return {
"tool.execute.after": async (input, output) => {
// Only shell-ish tools carry a command whose exit status is a verdict.
const command = commandOf(input);
if (!command || !TEST_COMMAND.test(command)) return;
const ok = !looksFailed(output);
const detail = `${command.slice(0, 120)}${ok ? " — passed" : " — failed"}`;
try {
const res = await fetch(`${ROUTER}/outcome`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ ok, detail, source: directory }),
// The router being down must never break the user's session.
signal: AbortSignal.timeout(3000),
});
if (res.status === 409) {
// Several sessions active and the directory did not match one the
// router had seen. Dropping the sample is the correct outcome.
console.error(
"[router-outcome] ambiguous session; outcome not recorded",
);
} else if (!res.ok && res.status !== 404) {
console.error(`[router-outcome] ${res.status} reporting outcome`);
}
} catch (err) {
// Swallowed on purpose: a reporting failure is not the user's problem.
console.error(`[router-outcome] could not reach ${ROUTER}: ${err.message}`);
}
},
};
};

56
docs/api.md
View File

@@ -140,6 +140,39 @@ structural verification verdict surfaces in the `X-Router-Verification` header
so a client can inspect it without parsing the response body. Valid values:
`ok`, `truncated`, `malformed`, `unverifiable`, `none`.
**Conversation identity headers** (optional): A client may attach three
headers to `POST /v1/chat/completions` so the router can group its decisions
by conversation and attribute an answer to the exact conversation that asked
for it. All three are optional; an absent or invalid header is treated as
absent (it is never an error).
| Header | Meaning | Length | Pattern |
|---|---|---|---|
| `X-Router-Conversation` | Id of ONE conversation (e.g. an opencode sessionID) | 1-128 chars | `^[A-Za-z0-9._:-]+$` |
| `X-Router-Agent` | Name of the agent making the request | 1-64 chars | `^[A-Za-z0-9._:-]+$` |
| `X-Router-Parent` | Parent conversation id, when this one is a sub-conversation | 1-128 chars | `^[A-Za-z0-9._:-]+$` |
Contract rules:
- Charset: letters, digits, `.`, `_`, `:`, `-` (the regex `^[A-Za-z0-9._:-]+$`);
anything else, or a length outside the table's range, makes the header
invalid and therefore absent.
- An absent or invalid header is treated as absent, never as an error: the
request is still routed normally.
- Namespacing: a valid `X-Router-Conversation` produces
`session_key = "c:" + conversation`, so client conversations are keyed
separately from the router's own hashed prompt fingerprints (which carry no
`c:` prefix). A valid `X-Router-Parent` produces `parent_key = "c:" + parent`.
Conversation ids never appear alongside message text: `session_key` stores
only the namespaced id, never the messages.
- When `X-Router-Conversation` is absent or invalid, the router falls back to
its hashed prompt fingerprint as `session_key` (see [routing.md](routing.md)
"incumbency and cache pricing").
These ids are what make `/outcome` attribution exact (below): an
`X-Router-Conversation` on the request sets the `session_key` row that a later
report can address by `conversation_id` without guessing `source` or ambiguity.
**Capability 422s**: When no model survives the hard filters, the 422 names the
active constraints. That now includes "vision-capable model" or
"json-mode-capable model" when the request carried images or a JSON-mode
@@ -162,6 +195,29 @@ client whose reports go nowhere finds out. The lookup checks
rows carry the same `request_id`/`session_dir` attribution) — cloud wins on a
rare id collision.
`conversation_id` is an optional extra body field that makes attribution
exact: it names the conversation (the same `X-Router-Conversation` the client
sent on the request) so a report can land without a `request_id`, and without
guessing which source the row came from.
```bash
# attribute a report to a conversation, no request_id needed
curl -s localhost:8080/outcome -H 'content-type: application/json' \
-d '{"conversation_id":"abc-123","ok":false,"detail":"tests failed"}'
```
Resolution order is strict:
1. `request_id` — exact id; unmatched returns `404`.
2. `conversation_id` (when present) — exact namespaced match on the
conversation's rows; unmatched returns `404` and never falls through.
3. `source` / unambiguous fallback — only when `conversation_id` is entirely
absent (no `source` given and exactly one plausible row).
When `conversation_id` is present but invalid (wrong charset or length) the
report is rejected with `422`, because a malformed id is a client bug, not an
absence.
Unlike the structural/local-LLM checks — which only ever record failures —
`/outcome` folds **both** directions into `proficiency` via `feedback.py`: a
`false` report counts against the model same as any other verification failure,

8
docs/data-model.md
View File

@@ -207,6 +207,7 @@ burn rate, projected runway, and stale-reading alerts.
| `call_type` | TEXT | Not a closed enum. Current values include `classify`, `verify`, `local_vision`, `local_dispatch`, `file_summarization`, `diff_checking`, and `seed_local_dispatch`. |
| `request_id` | TEXT | Optional; joins to `POST /outcome` reports the same way `energy_observations.request_id` does for cloud rows |
| `session_dir` | TEXT | Optional; used for source-less `/outcome` attribution |
| `session_key` | TEXT | Optional; namespaced conversation key (`"c:" + conversation`) from `X-Router-Conversation`, or NULL for a no-header local row |
| `avg_power_watts` | REAL | Averaged over the call (background `nvidia-smi` sampler) |
| `duration_seconds` | REAL | Wall-clock time for the local call |
| `energy_kwh` | REAL | `avg_power_watts × duration_seconds / 3_600_000` |
@@ -268,7 +269,9 @@ failure — it means the checker had nothing to say, not that the model failed.
| `est_cost_usd` | REAL | Estimated cost of the selected model |
| `est_proficiency` | REAL | Estimated proficiency for the task category |
| `rejected_reason` | TEXT | Active filters when nothing was selected |
| `session_key` | TEXT | Hashed session fingerprint ONLY |
| `session_key` | TEXT | `"c:" + conversation id` from `X-Router-Conversation` when the client sent one, else a hashed content fingerprint of the conversation; NULL on a no-header local row |
| `agent` | TEXT | Optional; agent name from `X-Router-Agent` (namespaced form, see [api.md](api.md)) |
| `parent_key` | TEXT | Optional; `"c:" + parent` `X-Router-Parent` conversation key, when the request named one |
| `tools` | INTEGER | 0/1 — request carried a `tools` array |
| `images` | INTEGER | 0/1 — request carried image parts |
| `json_mode` | INTEGER | 0/1 — request required JSON mode |
@@ -317,7 +320,8 @@ performing" is answerable: which model was picked, for what category/tier,
how long classification took, and — when nothing was selected — which hard
filter shut it out. It is an observability table: nothing in routing reads it.
It stores only a hashed session fingerprint in `session_key`; `session_dir`,
It keys a conversation by `session_key` -- `"c:" + conversation id` when the client
sent `X-Router-Conversation`, else a hashed content fingerprint; `session_dir`,
prompts, and answers are deliberately excluded. A test enforces that the write
path does not store prompt or answer text. The write is gated by
`logging.log_route_decisions` and is **best-effort**: a failed write is logged

7
docs/routing.md
View File

@@ -235,6 +235,13 @@ on a long prompt is most of the prompt. This is the first place the
router's own past decision feeds back into its cost model, and it is
deliberately narrow: it changes only the cost key, never the sort.
The incumbent is **per conversation** when the client sends
`X-Router-Conversation`: `session_key` is then the namespaced conversation id,
so each conversation tracks its own incumbent. Without that header the
incumbent is per system-prompt fingerprint, and the fingerprint's hashing
merges concurrent same-prompt agents into one incumbent so they share the
cache discount.
Four knobs under `objective:`, all shipping **off**
(`incumbent_cache_pricing: false`, dial `null`; enable only after the Wave 1
post-restart baseline day, the Wave 2 gate in `plans/token-waste-waves.md`):

15
src/config.py
View File

@@ -157,6 +157,12 @@ class Objective(StrictModel):
# cache rate is trusted for pricing decisions.
incumbent_rate_min_observations: int = 25
# Window (in seconds) for the conversation adoption counter in /metrics.
# Recent route_decisions rows within this window whose session_key starts
# with "c:" are counted. Null or absent means "all time" (no window);
# 0 is rejected by the adoption_window_positive validator below.
adoption_window_seconds: Optional[int] = None
credit_attenuation: CreditAttenuationConfig = CreditAttenuationConfig()
@field_validator("quality_tolerance")
@@ -367,6 +373,15 @@ class Objective(StrictModel):
raise ValueError("objective.incumbent_rate_min_observations must be >= 0")
return v
@field_validator("adoption_window_seconds")
@classmethod
def adoption_window_positive(cls, v: Optional[int]) -> Optional[int]:
if v is not None and v <= 0:
raise ValueError(
"objective.adoption_window_seconds must be > 0, or null/absent for all time"
)
return v
@model_validator(mode="after")
def _resolve_challenger_cache_rate(self) -> "Objective":
if self.incumbent_challenger_cache_rate is None:

139
src/conversation_identity.py Normal file
View File

@@ -0,0 +1,139 @@
"""Resolve conversation identity from client headers.
Pure module — no DB, no config, no dispatcher import. Like
``capabilities.py`` and ``exploration.py``, this module owns no I/O: headers
come in as a mapping and a resolved identity comes out as a value object.
Three optional client headers describe the conversation a request belongs to:
- ``X-Router-Conversation`` — the id of ONE conversation (1-128 chars).
- ``X-Router-Agent`` — the name of the calling agent (1-64 chars).
- ``X-Router-Parent`` — the conversation id of the parent (1-128 chars).
An absent or invalid header is treated as absent, never as an error. Each
valid value is prefixed with ``c:`` to build a namespaced key so a 16-hex
session fingerprint (which contains no colon) can never collide with a
conversation id.
Header names are matched case-insensitively, and when a name appears more than
once the FIRST occurrence wins — the module works identically whether handed a
``starlette.Headers`` (whose iteration yields lowercased names and whose
``.get()`` returns the first value) or a plain ``dict``. A later occurrence is
never consulted, even when the first value fails validation.
"""
from __future__ import annotations
import re
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Optional
# Allowed value charset for conversation/agent ids: alphanumerics, dot,
# underscore, colon and hyphen. Colon is allowed so a client MAY hand a value
# that is already prefixed; it is not required.
_VALUE_RE = re.compile(r"^[A-Za-z0-9._:-]+$")
_CONVERSATION_MAX = 128
_AGENT_MAX = 64
_CONVERSATION_HEADER = "x-router-conversation"
_AGENT_HEADER = "x-router-agent"
_PARENT_HEADER = "x-router-parent"
@dataclass(frozen=True)
class ConversationIdentity:
"""Resolved identity of a conversation from client headers.
``session_key`` is ``"c:" + conversation id`` when ``X-Router-Conversation``
is valid, else the caller-supplied session fingerprint (returned untouched).
``parent_key`` is ``"c:" + parent id`` when ``X-Router-Parent`` is valid,
else None. ``agent`` is the agent name when ``X-Router-Agent`` is valid,
else None.
"""
session_key: Optional[str]
agent: Optional[str]
parent_key: Optional[str]
def parse_conversation_id(value: object) -> Optional[str]:
"""Return ``value`` when it is a valid conversation/agent id, else None.
A valid id is a non-empty ``str`` of 1-128 characters matching
``^[A-Za-z0-9._:-]+$``. Anything else — a non-string, an empty string, a
value with a space/slash/newline or any other disallowed character, a value
longer than 128 characters — yields None.
"""
if not isinstance(value, str):
return None
if not value:
return None
if len(value) > _CONVERSATION_MAX:
return None
if not _VALUE_RE.fullmatch(value):
return None
return value
def _parse_agent(value: object) -> Optional[str]:
if not isinstance(value, str):
return None
if not value:
return None
if len(value) > _AGENT_MAX:
return None
if not _VALUE_RE.fullmatch(value):
return None
return value
def resolve_identity(
headers: Mapping[str, str], fingerprint: Optional[str]
) -> ConversationIdentity:
"""Resolve conversation identity from client headers.
``headers`` is any mapping of header name -> value. Header names are
matched case-insensitively. When a name appears more than once after
lowercasing, the FIRST occurrence wins — iteration over the mapping's items
in order preserves that across both a ``starlette.Headers`` (which yields
lowercased names) and a plain ``dict``. A later occurrence is never
consulted, even when the first value fails validation.
``fingerprint`` is the ``session_fingerprint(messages)`` value from the
caller (16 hex chars, no colon). It is used unchanged as ``session_key``
only when ``X-Router-Conversation`` is absent or invalid.
"""
conversation: Optional[str] = None
agent: Optional[str] = None
parent: Optional[str] = None
# First occurrence wins: once a lowered name has been seen, later
# occurrences are ignored even when the first value failed validation.
seen: set[str] = set()
for name, value in headers.items():
lowered = name.lower() if isinstance(name, str) else str(name)
if lowered in seen:
continue
seen.add(lowered)
if lowered == _CONVERSATION_HEADER:
conversation = parse_conversation_id(value)
elif lowered == _AGENT_HEADER:
agent = _parse_agent(value)
elif lowered == _PARENT_HEADER:
parent = parse_conversation_id(value)
session_key: Optional[str]
if conversation is not None:
session_key = "c:" + conversation
else:
session_key = fingerprint
parent_key: Optional[str] = "c:" + parent if parent is not None else None
return ConversationIdentity(
session_key=session_key,
agent=agent,
parent_key=parent_key,
)

158
src/dispatcher.py
View File

@@ -52,10 +52,10 @@ from typing import Any, Final, Literal, Optional
import requests
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, HTTPException
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse
from openai import APIStatusError, OpenAI, OpenAIError
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
import admin
import circuit_breaker
@@ -69,6 +69,7 @@ import session_cache
import textcut
from capabilities import detect_capabilities, iter_image_url_values
from config import BUILTIN_PROFILES, FlexPreference, RouterConfig, RoutingProfile, load_config
from conversation_identity import parse_conversation_id, resolve_identity
from context_prune import (
_text_only,
estimate_tokens,
@@ -81,6 +82,7 @@ from iteration import attempts_allowed, plan_retry
from metrics import (
cache_rate_series,
cache_rate_warnings,
conversation_adoption,
cost_estimate_calibration,
latency_series,
local_energy_summary,
@@ -450,7 +452,9 @@ def ensure_route_decisions(conn: sqlite3.Connection) -> None:
profile TEXT,
prefix_divergence_index INTEGER,
prefix_tokens_after_divergence INTEGER,
prefix_prev_message_count INTEGER
prefix_prev_message_count INTEGER,
agent TEXT,
parent_key TEXT
)
"""
)
@@ -488,6 +492,11 @@ def ensure_route_decisions(conn: sqlite3.Connection) -> None:
("prefix_divergence_index", "INTEGER"),
("prefix_tokens_after_divergence", "INTEGER"),
("prefix_prev_message_count", "INTEGER"),
# Conversation identity. Additive and NULL on every existing row;
# a row written before these columns existed cannot name the agent
# that decided it or link to a parent decision.
("agent", "TEXT"),
("parent_key", "TEXT"),
):
if name not in existing:
conn.execute(f"ALTER TABLE route_decisions ADD COLUMN {name} {decl}")
@@ -539,6 +548,7 @@ def _ensure_tables() -> None:
try:
ensure_route_decisions(conn)
_ensure_energy_observations_table(conn)
local_energy.ensure_local_energy_table(conn)
admin.ensure_admin_tables(conn)
finally:
conn.close()
@@ -1459,13 +1469,15 @@ def _log_local_energy(
*,
request_id: Optional[str] = None,
session_dir: Optional[str] = None,
session_key: Optional[str] = None,
) -> None:
"""Persist one local-energy observation computed from a measurement.
The caller must already have checked that metering is enabled and the call
site is loopback; this helper just does the arithmetic and the insert.
request_id and session_dir are forwarded so POST /outcome can attribute
local-dispatch answers via the local energy ledger.
local-dispatch answers via the local energy ledger; session_key records
the conversation the local answer belongs to.
"""
avg_power_watts = measurement.avg_power_watts
duration_seconds = measurement.duration_seconds
@@ -1491,6 +1503,7 @@ def _log_local_energy(
observed_at=datetime.now(timezone.utc).isoformat(),
request_id=request_id,
session_dir=session_dir,
session_key=session_key,
)
finally:
conn.close()
@@ -2021,6 +2034,8 @@ def persist_route_decision(
pinch_final_tokens=None,
profile: Optional[str] = None,
prefix_divergence: Optional[prefix_probe.Divergence] = None,
agent: Optional[str] = None,
parent_key: Optional[str] = None,
) -> Optional[int]:
"""Record one routing decision to route_decisions, best-effort and gated.
@@ -2130,8 +2145,8 @@ def persist_route_decision(
flex_preference, flex_swapped, flex_forced, exploration,
request_id, pinch_original_tokens, pinch_final_tokens, profile,
prefix_divergence_index, prefix_tokens_after_divergence,
prefix_prev_message_count
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
prefix_prev_message_count, agent, parent_key
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
observed_at,
@@ -2166,6 +2181,8 @@ def persist_route_decision(
prefix_index,
prefix_tokens_after,
prefix_prev_count,
agent,
parent_key,
),
)
decision_id: Optional[int] = int(cursor.lastrowid)
@@ -2209,6 +2226,8 @@ def persist_route_decision(
"prefix_divergence_index": prefix_index,
"prefix_tokens_after_divergence": prefix_tokens_after,
"prefix_prev_message_count": prefix_prev_count,
"agent": agent,
"parent_key": parent_key,
}
)
return decision_id
@@ -2643,6 +2662,7 @@ def metrics_endpoint():
"top_proficiency": top_proficiency(conn, "coding_general"),
"local_energy": local_energy_summary(conn, cfg),
"pinch": pinch_summary(conn, cfg),
"adoption": conversation_adoption(conn, cfg),
# Two report-only series. Top-level rather than inside `coverage`
# because neither answers a coverage question, and deliberately
# NOT read by anything on the routing path -- see their docstrings
@@ -2735,6 +2755,30 @@ class OutcomeReport(BaseModel):
source: Optional[str] = Field(
None, description="Where the report came from, e.g. the project directory."
)
conversation_id: Optional[str] = Field(
None,
description=(
"The conversation this report belongs to, when the client knows it. "
"An exact conversation resolves a report even without request_id."
),
)
@field_validator("conversation_id")
@classmethod
def _validate_conversation_id(cls, v: object) -> Optional[str]:
"""Validate an explicitly-provided conversation_id.
``parse_conversation_id`` returns None for ANY invalid or absent value,
but an absent field and a present-but-invalid one are different: the
caller may omit conversation_id deliberately (to fall back to source),
so only a present-but-invalid value is an error. An invalid value is a
client bug and FastAPI surfaces the ValueError as a 422.
"""
if v is None:
return None
if parse_conversation_id(v) is None:
raise ValueError("conversation_id must match ^[A-Za-z0-9._:-]+$ (1-128 chars)")
return parse_conversation_id(v)
AMBIGUOUS = object()
@@ -2789,7 +2833,7 @@ def _most_recent_if_unambiguous(conn: sqlite3.Connection):
recent_local = conn.execute(
f"""
SELECT id, request_id, model_id, 'ollama-local' AS provider,
call_type AS task_category, session_dir AS session_key
call_type AS task_category, COALESCE(session_key, session_dir) AS session_key
FROM local_energy_observations
WHERE request_id IS NOT NULL
AND call_type != ?
@@ -2861,15 +2905,19 @@ def _find_outcome_row(
conn: sqlite3.Connection,
request_id: Optional[str],
source: Optional[str],
conversation_id: Optional[str] = None,
):
"""Resolve an outcome report to a normalized observation row.
Checks energy_observations first (cloud rows win on request_id
collisions), then local_energy_observations for local-dispatch answers.
Contract rule 4: ``request_id`` wins first; else, when ``conversation_id``
is present, the most recent row keyed by that conversation in
energy_observations, then local_energy_observations; else, only when
``conversation_id`` is ABSENT, ``source`` and the recent-session fallback
run as before.
Returns a dict-like sqlite3.Row with keys request_id, model_id, provider,
task_category. Returns AMBIGUOUS when the fallback recent-session lookup
cannot confidently pick one conversation; returns None when nothing
matches.
task_category. Returns AMBIGUOUS when the recent-session lookup cannot
confidently pick one conversation; returns None when nothing matches.
"""
if request_id:
row = conn.execute(
@@ -2895,6 +2943,36 @@ def _find_outcome_row(
return row
return None
if conversation_id is not None:
# Exact conversation. Unmatched must not fall through to source: a
# client that named its conversation explicitly wants THAT one or a 404.
key = "c:" + conversation_id
row = conn.execute(
"""
SELECT id, request_id, model_id, provider, task_category
FROM energy_observations
WHERE session_key = ? AND request_id IS NOT NULL
AND task_category != ?
ORDER BY id DESC LIMIT 1
""",
(key, SEED_CATEGORY),
).fetchone()
if row is None:
row = conn.execute(
"""
SELECT id, request_id, model_id, 'ollama-local' AS provider,
call_type AS task_category
FROM local_energy_observations
WHERE session_key = ? AND request_id IS NOT NULL
AND call_type != ?
ORDER BY id DESC LIMIT 1
""",
(key, SEED_CATEGORY),
).fetchone()
if row is not None:
return row
return None
if source:
# The client told us where it is. If any completion came from a
# conversation naming that directory, this is exact even with
@@ -2954,7 +3032,9 @@ def report_outcome(report: OutcomeReport):
logs.new_trace()
conn = _db()
try:
row = _find_outcome_row(conn, report.request_id, report.source)
row = _find_outcome_row(
conn, report.request_id, report.source, report.conversation_id
)
finally:
conn.close()
@@ -2983,14 +3063,23 @@ def report_outcome(report: OutcomeReport):
)
# 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.",
)
if report.conversation_id is not None:
message = (
f"No routed completion found for conversation "
f"{report.conversation_id!r}. Only completions this router "
"dispatched can be reported on."
)
elif report.request_id:
message = (
f"No routed completion found for request_id {report.request_id!r}. "
"Only completions this router dispatched can be reported on."
)
else:
message = (
"No completions have been routed yet, so there is nothing to "
"report on."
)
raise HTTPException(404, message)
attributable = _outcome_is_attributable(conn, row["request_id"])
log_verification(
@@ -3675,6 +3764,7 @@ def _run_local_dispatch(
*,
category: Optional[str],
body: dict,
session_key: Optional[str] = None,
) -> dict:
"""Dispatch to a local (Ollama-compatible) model via the OpenAI compat endpoint.
@@ -3786,6 +3876,7 @@ def _run_local_dispatch(
measurement=ctx,
request_id=request_id,
session_dir=session_dir,
session_key=session_key,
)
if cfg.circuit_breaker.enabled:
circuit_breaker.record_failure(
@@ -3824,6 +3915,7 @@ def _run_local_dispatch(
measurement=ctx,
request_id=request_id,
session_dir=session_dir,
session_key=session_key,
)
raise HTTPException(
status_code=502,
@@ -3854,6 +3946,7 @@ def _run_local_dispatch(
measurement=ctx,
request_id=request_id,
session_dir=session_dir,
session_key=session_key,
)
raise HTTPException(
status_code=502,
@@ -3881,6 +3974,7 @@ def _run_local_dispatch(
measurement=ctx,
request_id=request_id,
session_dir=session_dir,
session_key=session_key,
)
raise HTTPException(
status_code=502,
@@ -3899,6 +3993,7 @@ def _run_local_dispatch(
measurement=ctx,
request_id=request_id,
session_dir=session_dir,
session_key=session_key,
)
return {
@@ -4134,7 +4229,7 @@ def _sniff_telemetry_line(line: str) -> Optional[tuple[str, dict]]:
@app.post("/v1/chat/completions")
def chat_completions(body: dict[str, Any], background: BackgroundTasks):
def chat_completions(body: dict[str, Any], background: BackgroundTasks, request: Request):
"""OpenAI-compatible completions, routed then proxied.
Streaming is passed through chunk by chunk rather than buffered, so a
@@ -4150,7 +4245,13 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
# 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)
fingerprint = session_fingerprint(messages)
# A client conversation header overrides the content fingerprint when
# present and valid; agent/parent are recorded for attribution. The
# resolved value is threaded explicitly (never via a ContextVar) to the
# persist sites so it survives stream-generator frames after the handler.
identity = resolve_identity(request.headers, fingerprint)
session_key = identity.session_key
# Publish for the classifier fallback cascade, which needs it two
# frames down without route() having to carry it.
_current_session_key.set(session_key)
@@ -4372,6 +4473,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
selected_provider="local",
rejected_reason=None,
session_key=session_key,
agent=identity.agent, parent_key=identity.parent_key,
tools=tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
@@ -4412,6 +4514,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
classification=decision,
rejected_reason="; ".join(limits),
session_key=session_key,
agent=identity.agent, parent_key=identity.parent_key,
tools=tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
@@ -4439,6 +4542,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
"chat",
classification=decision,
session_key=session_key,
agent=identity.agent, parent_key=identity.parent_key,
tools=tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
@@ -4533,6 +4637,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
selected_provider=provider,
rejected_reason=None,
session_key=session_key,
agent=identity.agent, parent_key=identity.parent_key,
tools=int(caps.tools_present),
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),
@@ -4564,7 +4669,8 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
"row selected but no matching local_dispatch_models entry",
)
result = _run_local_dispatch(
entry, send_messages, category=category or "local_dispatch", body=body
entry, send_messages, category=category or "local_dispatch", body=body,
session_key=session_key,
)
return _local_dispatch_response(
result["payload"], entry, streaming=streamed, request_id=result["request_id"]
@@ -4640,7 +4746,8 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
)
try:
result = _run_local_dispatch(
entry, send_messages, category=category, body=body
entry, send_messages, category=category, body=body,
session_key=session_key,
)
except HTTPException as exc:
logs.warning(
@@ -4656,6 +4763,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
selected_provider="ollama-local",
rejected_reason=f"cloud_failed:{cloud_model}:{cloud_status}",
session_key=session_key,
agent=identity.agent, parent_key=identity.parent_key,
tools=caps.tools_present,
images=int(caps.has_images),
json_mode=int(caps.require_json_mode),

11
src/local_energy.py
View File

@@ -142,6 +142,7 @@ _TABLE_COLUMNS = [
("call_type", "TEXT NOT NULL"),
("request_id", "TEXT"),
("session_dir", "TEXT"),
("session_key", "TEXT"),
("avg_power_watts", "REAL"),
("duration_seconds", "REAL"),
("energy_kwh", "REAL"),
@@ -154,6 +155,7 @@ _TABLE_COLUMNS = [
_ADDITIVE_COLUMNS = [
("request_id", "TEXT"),
("session_dir", "TEXT"),
("session_key", "TEXT"),
]
@@ -198,22 +200,24 @@ def log_local_energy(
*,
request_id: Optional[str] = None,
session_dir: Optional[str] = None,
session_key: Optional[str] = None,
) -> None:
"""Insert a row into ``local_energy_observations``.
The caller computes energy, cost, and carbon; this helper only persists
them. ``ensure_local_energy_table`` is called first so the write succeeds
even on older databases. request_id/session_dir are keyword-only so
existing positional call sites keep working unmodified.
even on older databases. request_id/session_dir/session_key are keyword-only
so existing positional call sites keep working unmodified.
"""
ensure_local_energy_table(conn)
conn.execute(
"""
INSERT INTO local_energy_observations (
observed_at, model_id, call_type, request_id, session_dir,
session_key,
avg_power_watts, duration_seconds, energy_kwh, cost_usd,
carbon_g_co2eq, meter
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
observed_at,
@@ -221,6 +225,7 @@ def log_local_energy(
call_type,
request_id,
session_dir,
session_key,
avg_power_watts,
duration_seconds,
energy_kwh,

97
src/metrics.py
View File

@@ -2927,6 +2927,14 @@ _PREFIX_PROBE_COLUMNS: Final = (
"prefix_prev_message_count",
)
# Conversation-identity columns (agent, parent_key) that arrive by ALTER at
# dispatcher start-up. Probed and backfilled the same way as the prefix-probe
# columns above so an un-migrated DB still loads.
_CONVERSATION_IDENTITY_COLUMNS: Final = (
"agent",
"parent_key",
)
def recent_decisions(
conn: sqlite3.Connection,
@@ -2948,7 +2956,13 @@ def recent_decisions(
for column in _PREFIX_PROBE_COLUMNS
if _has_column(conn, "route_decisions", column)
]
identity_present = [
column
for column in _CONVERSATION_IDENTITY_COLUMNS
if _has_column(conn, "route_decisions", column)
]
probe_select = "".join(f", {column}" for column in present)
identity_select = "".join(f", {column}" for column in identity_present)
rows = [
dict(row)
for row in conn.execute(
@@ -2961,7 +2975,7 @@ def recent_decisions(
rejected_reason, session_key, tools, images, json_mode, streamed,
flex_preference, flex_swapped, flex_forced,
exploration, request_id,
pinch_original_tokens, pinch_final_tokens, profile{probe_select}
pinch_original_tokens, pinch_final_tokens, profile{probe_select}{identity_select}
FROM route_decisions
ORDER BY id DESC
LIMIT ?
@@ -2975,9 +2989,90 @@ def recent_decisions(
for row in rows:
for column in _PREFIX_PROBE_COLUMNS:
row.setdefault(column, None)
for column in _CONVERSATION_IDENTITY_COLUMNS:
row.setdefault(column, None)
return rows
def conversation_adoption(
conn: sqlite3.Connection,
cfg: Any,
) -> dict:
"""Conversation-adoption counters over route_decisions.
This is the (H1) adoption metric: of the routing decisions in the
window, how many carry a proper conversation session key (``c:...``)
rather than a simple fingerprint hash. Returns:
* ``n_c_conversations`` - COUNT(DISTINCT session_key) over the rows
whose session_key starts with ``c:`` (one per conversation).
* ``n_c_decisions`` - the row count over those same rows (one per turn;
a former version named this row count ``n_c_conversations``, which
over-counted multi-turn conversations).
* ``n_decisions`` - all route_decisions rows in the same window,
fingerprint rows included; this denominates ``share``.
* ``share`` - ``n_c_decisions / n_decisions``, or None when the window
holds no decisions at all.
The window is ``objective.adoption_window_seconds``; null or absent
means all time; 0 is rejected by the config validator. All three counts
read the same window, so ``share`` is a true in-window ratio.
The capability gate checks ``route_decisions.session_key`` only (PRAGMA
table_info): a database that has not run the identity migration yet
yields zeros, not an error.
SQLite LIKE is case-insensitive by default, fine here because the ``c:``
prefix is the only non-c delimiter and fingerprints never contain a colon.
Switch to ``LIKE 'c:%' ESCAPE '\'`` only if a future prefix needs escaping.
"""
window_seconds = getattr(cfg.objective, "adoption_window_seconds", None)
if not _has_column(conn, "route_decisions", "session_key"):
return {
"n_c_conversations": 0,
"n_c_decisions": 0,
"n_decisions": 0,
"share": None,
"window_seconds": window_seconds,
}
params: list = []
where_clause = "1 = 1"
if window_seconds and window_seconds > 0:
where_clause = (
"julianday(observed_at) > "
"julianday('now', '-' || ? || ' seconds')"
)
params.append(str(window_seconds))
sql = f"""
SELECT SUM(CASE WHEN session_key LIKE 'c:%' THEN 1 ELSE 0 END)
AS n_c_decisions,
COUNT(DISTINCT CASE WHEN session_key LIKE 'c:%'
THEN session_key END)
AS n_c_conversations,
COUNT(*) AS n_decisions
FROM route_decisions
WHERE {where_clause}
"""
if params:
row = conn.execute(sql, tuple(params)).fetchone()
else:
row = conn.execute(sql).fetchone()
# A no-GROUP-BY aggregate SELECT always yields exactly one row, but SUM
# over that empty set is NULL in SQLite -- coerce it, COUNT never is.
n_c_decisions = row["n_c_decisions"] or 0
n_decisions = row["n_decisions"]
return {
"n_c_conversations": row["n_c_conversations"],
"n_c_decisions": n_c_decisions,
"n_decisions": n_decisions,
"share": (n_c_decisions / n_decisions) if n_decisions else None,
"window_seconds": window_seconds,
}
def per_model(conn: sqlite3.Connection) -> List[dict]:
"""Per-model aggregates over the last 30 d of energy_observations."""
return [

1
src/tui_model.py
View File

@@ -248,6 +248,7 @@ def decision_row(r: dict) -> dict:
"prefix_prev_message_count": r.get("prefix_prev_message_count"),
"request_id": r.get("request_id"),
"session_key": r.get("session_key"),
"agent": r.get("agent"),
}

34
tests/fixtures/route_decision_no_header.json vendored Normal file
View File

@@ -0,0 +1,34 @@
{
"agent": null,
"candidates_considered": 2,
"classification_source": "classifier",
"confidence": 0.9,
"est_cost_usd": 0.00016,
"est_proficiency": 0.5,
"exploration": 0,
"flex_forced": 0,
"flex_preference": "auto",
"flex_swapped": 0,
"images": 0,
"json_mode": 0,
"kind": "chat",
"latency_tolerance": "interactive",
"parent_key": null,
"pinch_final_tokens": 5,
"pinch_original_tokens": 5,
"prefix_divergence_index": null,
"prefix_prev_message_count": null,
"prefix_tokens_after_divergence": null,
"profile": "default",
"rejected_reason": null,
"request_id": "chatcmpl-test",
"required_context_tokens": 100,
"runner_up_models": "[{\"model_id\": \"dear-model\", \"provider\": \"neuralwatt\"}]",
"selected_model": "cheap-model",
"selected_provider": "neuralwatt",
"session_key": "be41d1308beb018d",
"streamed": 0,
"task_category": "coding_general",
"task_tier": 2,
"tools": 0
}

4
tests/test_admin_knob_coverage.py
View File

@@ -320,6 +320,10 @@ DELIBERATELY_NOT_IN_ADMIN: dict[str, str] = {
"sample floor for the latency percentiles, applied separately to the "
"wall and TTFT counts; a statistical guard on a report."
),
"objective.adoption_window_seconds": (
"lookback for the conversation adoption counter in /metrics; a "
"read-only window that shapes a report, not a routing dial."
),
}

16
tests/test_config_endpoints.py
View File

@@ -352,6 +352,22 @@ def test_billing_reset_day_rejects_out_of_range(raw):
RouterConfig(**cfg)
def test_adoption_window_seconds_rejects_zero():
"""0 is rejected by adoption_window_positive; null or absent means all time."""
from config import Objective
with pytest.raises(ValueError, match="adoption_window_seconds"):
Objective(adoption_window_seconds=0)
def test_adoption_window_seconds_null_is_accepted():
"""None is the documented all-time value."""
from config import Objective
obj = Objective(adoption_window_seconds=None)
assert obj.adoption_window_seconds is None
def test_a_removed_key_is_rejected_rather_than_ignored(raw):
"""log_path named a file nothing ever wrote; leaving it valid would lie."""
import copy

188
tests/test_conversation_identity.py Normal file
View File

@@ -0,0 +1,188 @@
"""Tests for the pure conversation-identity resolver.
All tests are offline; no provider, local-model, or network calls are made.
The duplicate-header tests use a real ``starlette.Headers`` to prove the
module behaves identically across both header container shapes.
"""
from starlette.datastructures import Headers
from conversation_identity import (
ConversationIdentity,
parse_conversation_id,
resolve_identity,
)
class TestParseConversationId:
def test_valid_id_returns_value(self) -> None:
assert parse_conversation_id("abc-123._:XYZ") == "abc-123._:XYZ"
def test_valid_single_char(self) -> None:
assert parse_conversation_id("a") == "a"
def test_valid_max_length_128(self) -> None:
value = "a" * 128
assert parse_conversation_id(value) == value
def test_129_char_id_ignored(self) -> None:
assert parse_conversation_id("a" * 129) is None
def test_space_ignored(self) -> None:
assert parse_conversation_id("abc def") is None
def test_slash_ignored(self) -> None:
assert parse_conversation_id("abc/def") is None
def test_newline_ignored(self) -> None:
assert parse_conversation_id("abc\ndef") is None
def test_empty_string_ignored(self) -> None:
assert parse_conversation_id("") is None
def test_non_string_ignored(self) -> None:
assert parse_conversation_id(123) is None
assert parse_conversation_id(None) is None
assert parse_conversation_id(b"abc") is None
class TestResolveIdentitySessionKey:
def test_valid_conversation_gives_c_prefixed_key(self) -> None:
ident = resolve_identity({"X-Router-Conversation": "conv1"}, "fp1234567890abcdef")
assert ident.session_key == "c:conv1"
def test_uppercase_header_name_accepted(self) -> None:
ident = resolve_identity({"X-ROUTER-CONVERSATION": "conv1"}, None)
assert ident.session_key == "c:conv1"
def test_header_with_any_case_accepted(self) -> None:
for name in (
"x-router-conversation",
"X-Router-Conversation",
"X-ROUTER-CONVERSATION",
):
ident = resolve_identity({name: "conv1"}, None)
assert ident.session_key == "c:conv1"
def test_fingerprint_fallback_returned_untouched(self) -> None:
fp = "0123456789abcdef"
ident = resolve_identity({}, fp)
assert ident.session_key is fp
def test_invalid_conversation_falls_back_to_fingerprint(self) -> None:
fp = "0123456789abcdef"
ident = resolve_identity({"X-Router-Conversation": "bad value"}, fp)
assert ident.session_key is fp
def test_fingerprint_shaped_value_still_gets_c_prefix(self) -> None:
# A 16-hex value passed as a conversation id is a namespace check: it
# must get the c: prefix so it cannot collide with an unfingerprinted
# session key.
ident = resolve_identity({"X-Router-Conversation": "0123456789abcdef"}, None)
assert ident.session_key == "c:0123456789abcdef"
def test_fingerprint_none_when_no_conversation_and_no_fingerprint(self) -> None:
ident = resolve_identity({}, None)
assert ident.session_key is None
class TestResolveIdentityAgent:
def test_valid_agent(self) -> None:
ident = resolve_identity({"X-Router-Agent": "my-agent_2"}, None)
assert ident.agent == "my-agent_2"
def test_agent_absent(self) -> None:
ident = resolve_identity({}, None)
assert ident.agent is None
def test_agent_65_chars_ignored(self) -> None:
ident = resolve_identity({"X-Router-Agent": "a" * 65}, None)
assert ident.agent is None
def test_agent_64_chars_accepted(self) -> None:
ident = resolve_identity({"X-Router-Agent": "a" * 64}, None)
assert ident.agent == "a" * 64
def test_agent_with_disallowed_char_ignored(self) -> None:
ident = resolve_identity({"X-Router-Agent": "my agent"}, None)
assert ident.agent is None
class TestResolveIdentityParent:
def test_valid_parent_gives_c_prefixed_key(self) -> None:
ident = resolve_identity({"X-Router-Parent": "parent1"}, None)
assert ident.parent_key == "c:parent1"
def test_parent_absent(self) -> None:
ident = resolve_identity({}, None)
assert ident.parent_key is None
def test_invalid_parent_ignored(self) -> None:
ident = resolve_identity({"X-Router-Parent": "has space"}, None)
assert ident.parent_key is None
class TestDuplicateHeaders:
def test_starlette_headers_first_wins(self) -> None:
# starlette.Headers yields lowercased names and .get() returns the
# first value; the module must read the first occurrence.
headers = Headers(
{
"X-Router-Conversation": "first",
"x-router-conversation": "second",
}
)
ident = resolve_identity(headers, None)
assert ident.session_key == "c:first"
def test_plain_dict_lowercase_collision_first_wins(self) -> None:
headers = {
"X-Router-Conversation": "A",
"x-router-conversation": "B",
}
ident = resolve_identity(headers, None)
assert ident.session_key == "c:A"
def test_starlette_headers_invalid_first_not_replaced_by_second(self) -> None:
# The FIRST occurrence wins even when its value is invalid: a later,
# valid occurrence must not be adopted.
headers = Headers(
{
"X-Router-Conversation": "bad value",
"x-router-conversation": "good",
}
)
fp = "0123456789abcdef"
ident = resolve_identity(headers, fp)
assert ident.session_key is fp
def test_plain_dict_invalid_first_wins_over_valid_second(self) -> None:
headers = {
"X-Router-Conversation": "bad value",
"x-router-conversation": "good",
}
fp = "0123456789abcdef"
ident = resolve_identity(headers, fp)
assert ident.session_key is fp
def test_agent_invalid_first_not_replaced_by_second(self) -> None:
headers = Headers(
{
"X-Router-Agent": "has space",
"x-router-agent": "good-agent",
}
)
ident = resolve_identity(headers, None)
assert ident.agent is None
class TestDataclassShape:
def test_exposes_exactly_expected_fields(self) -> None:
assert list(ConversationIdentity.__dataclass_fields__.keys()) == [
"session_key",
"agent",
"parent_key",
]
def test_is_frozen(self) -> None:
assert ConversationIdentity.__dataclass_params__.frozen is True

346
tests/test_conversation_wiring.py Normal file
View File

@@ -0,0 +1,346 @@
"""Task 3 acceptance: conversation identity wired into chat_completions.
These tests exercise the integration the wiring added end to end: a client
``X-Router-Conversation`` header must scope the recorded ``session_key`` (and
therefore the session incumbent, the cloud ``energy_observations`` ledger and
the local-dispatch energy ledger), while ``X-Router-Agent``/``X-Router-Parent``
must land in their columns. A headerless request must still behave exactly as
before — one fingerprint key, NULL agent/parent, byte-identical to the frozen
pre-wiring snapshot (``tests/fixtures/route_decision_no_header.json``).
The commit that added the snapshot fixture deliberately precedes this file, so
a clean checkout can read it without a gitignored ``.omo/evidence/`` path.
"""
import sqlite3
from types import SimpleNamespace
import pytest
import dispatcher
from test_no_header_snapshot import (
assert_matches_snapshot,
make_router,
no_header_payload,
)
CHEAP = "cheap-model"
DEAR = "dear-model"
@pytest.fixture(autouse=True, scope="module")
def _clear_prefix_probe_state():
"""Clean up ``prefix_probe._store`` after our module so the snapshot
test (runs alphabetically after us) sees a clean prefix-probe cache.
The probe stores per-session fingerprints in process-memory and a
POST with the snapshot's payload here would otherwise make a later
test see a prior turn and report non-NULL prefix divergence."""
yield
import prefix_probe
prefix_probe._store.clear()
_SYS = "You work in /home/user/proj and follow the design docs."
def _messages(*roles):
return [
{"role": r, "content": c}
for r, c in zip(roles, ["system prompt", "user query one", "assistant reply", "user query two"])
][: len(roles)]
def _rows(db_path, table, where=None, order="id"):
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
q = f"SELECT * FROM {table}"
if where:
q += f" WHERE {where}"
q += f" ORDER BY {order}"
rows = [dict(r) for r in conn.execute(q).fetchall()]
conn.close()
return rows
def _latest_decision(db_path):
rows = _rows(db_path, "route_decisions")
assert rows, "no route_decisions row"
return rows[-1]
def _post(client, *, headers=None, model="auto", msgs=None):
payload = {
"model": model,
"messages": msgs if msgs is not None else _messages("system", "user"),
}
return client.post(
"/v1/chat/completions", json=payload, headers=headers or {}
)
# --- Acceptance 1: same prompt, different conversation -> distinct keys -----
def test_same_system_prompt_distinct_conversations_get_distinct_session_keys(
tmp_path, monkeypatch
):
client, db_path = make_router(tmp_path, monkeypatch)
msgs = _messages("system", "user")
r1 = _post(client, headers={"X-Router-Conversation": "a"}, msgs=msgs)
r2 = _post(client, headers={"X-Router-Conversation": "b"}, msgs=msgs)
assert r1.status_code == 200 and r2.status_code == 200
keys = {r["session_key"] for r in _rows(db_path, "route_decisions")}
assert keys == {"c:a", "c:b"}, keys
# --- Acceptance 2: each conversation's incumbent is its OWN last model -----
def test_incumbent_is_conversation_scoped_not_global(
tmp_path, monkeypatch
):
client, db_path = make_router(tmp_path, monkeypatch)
conn = sqlite3.connect(db_path)
for sk, model in (
("c:a", CHEAP), # interleave so the GLOBAL last chat row is cheap
("c:b", DEAR),
("c:a", CHEAP),
):
conn.execute(
"""INSERT INTO route_decisions (
observed_at, kind, session_key, selected_provider,
selected_model, task_category, task_tier
) VALUES ('2026-09-13T00:00:00+00:00', 'chat', ?, 'neuralwatt',
?, 'coding_general', 2)""",
(sk, model),
)
conn.commit()
conn.close()
monkeypatch.setattr(
dispatcher.cfg.objective, "incumbent_cache_pricing", True
)
captured = {}
class Capturer:
def __call__(self, rows, **kwargs):
captured["incumbent"] = kwargs.get("incumbent")
from routing import rank_candidates as _real
return _real(rows, **kwargs)
monkeypatch.setattr(dispatcher, "rank_candidates", Capturer())
r = _post(client, headers={"X-Router-Conversation": "b"})
assert r.status_code == 200, r.text
# Conversation b's own last chat row is dear-model, even though the
# overall-last chat row belongs to conversation a (cheap-model).
assert captured["incumbent"] == ("neuralwatt", DEAR), captured["incumbent"]
# --- Acceptance 3: headerless requests land under one fingerprint, NULL
# agent/parent, byte-identical to the committed snapshot -----------------
def test_headerless_requests_reproduce_committed_snapshot(
tmp_path, monkeypatch
):
client, db_path = make_router(tmp_path, monkeypatch)
# The exact payload the committed snapshot was frozen from: the row must
# be byte-identical, so the messages (and therefore the fingerprint,
# pinch and prefix-probe values) must match the fixture's.
payload = no_header_payload()
# First POST: fresh session, no prior prefix-probe state → the row has
# NULL prefix columns and is byte-identical to the pre-wiring fixture.
r1 = _post(client, msgs=payload["messages"])
assert r1.status_code == 200, r1.text
assert_matches_snapshot(db_path)
# Second POST in the same session: the prefix probe now sees a prior
# turn and writes non-NULL prefix columns, so the byte-identical
# assertion cannot include this row. But both rows must share one
# fingerprint key and carry NULL agent/parent.
r2 = _post(client, msgs=payload["messages"])
assert r2.status_code == 200, r2.text
rows = _rows(db_path, "route_decisions")
assert len(rows) == 2
assert rows[0]["session_key"] == rows[1]["session_key"]
assert rows[0]["session_key"] is not None
for r in rows:
assert r["agent"] is None
assert r["parent_key"] is None
# --- Acceptance 4: agent and parent land in their columns -------------------
def test_agent_and_parent_land_in_columns(tmp_path, monkeypatch):
client, db_path = make_router(tmp_path, monkeypatch)
r = _post(
client,
headers={
"X-Router-Conversation": "conv1",
"X-Router-Agent": "my-agent",
"X-Router-Parent": "parent1",
},
)
assert r.status_code == 200, r.text
row = _latest_decision(db_path)
assert row["session_key"] == "c:conv1"
assert row["agent"] == "my-agent"
assert row["parent_key"] == "c:parent1"
# --- Acceptance 5: invalid conversation header falls back to fingerprint ----
def test_invalid_conversation_header_falls_back_to_fingerprint(
tmp_path, monkeypatch
):
client, db_path = make_router(tmp_path, monkeypatch)
msgs = _messages("system", "user")
bad = _post(
client,
headers={"X-Router-Conversation": "has space and /slash"},
msgs=msgs,
)
assert bad.status_code == 200, bad.text
good = _post(client, msgs=msgs)
assert good.status_code == 200, good.text
rows = _rows(db_path, "route_decisions")
bad_key, fp_key = rows[0]["session_key"], rows[1]["session_key"]
# The invalid header is treated as absent: same fingerprint as no header.
assert bad_key == fp_key
assert not bad_key.startswith("c:")
# --- Acceptance 6: energy ledgers get the conversation key ------------------
def test_cloud_energy_session_key_starts_with_c(tmp_path, monkeypatch):
client, db_path = make_router(tmp_path, monkeypatch)
r = _post(
client,
headers={"X-Router-Conversation": "ener"},
model="auto",
msgs=_messages("system", "user"),
)
assert r.status_code == 200, r.text
rows = _rows(db_path, "energy_observations")
assert rows, "expected an energy_observations row"
for row in rows:
assert (row["session_key"] or "").startswith("c:"), row
def _local_dispatch_harness(tmp_path, monkeypatch, session_key):
"""Drive one real _run_local_dispatch and return the produced row."""
client, db_path = make_router(tmp_path, monkeypatch)
model_id = "local-q"
monkeypatch.setattr(dispatcher.cfg.local_energy, "enabled", True)
# local_energy_dispatch_models is a read-only cached property on the
# Pydantic config; shadow it on the class so the metering gate sees our
# model as meterable.
monkeypatch.setattr(
type(dispatcher.cfg), "local_energy_dispatch_models",
property(lambda self: frozenset({model_id})),
)
monkeypatch.setattr(dispatcher.cfg.circuit_breaker, "enabled", False)
class FakeMeasurement:
avg_power_watts = 100.0
duration_seconds = 2.0
def __enter__(self):
return self
def __exit__(self, *args):
return False
monkeypatch.setattr(
dispatcher.local_energy, "measure",
lambda *a, **k: FakeMeasurement(),
)
def fake_post(url, headers=None, json=None, stream=False, timeout=None):
resp = SimpleNamespace()
resp.status_code = 200
resp.json = lambda: {
"id": "ollama-local-1",
"choices": [
{"message": {"role": "assistant", "content": "local answer"}}
],
"usage": {"prompt_tokens": 20, "completion_tokens": 8},
}
resp.text = "{}"
return resp
monkeypatch.setattr(dispatcher.requests, "post", fake_post)
entry = SimpleNamespace(
model_id=model_id,
base_url="http://localhost:11434/v1",
api_key_env=None,
max_output_tokens=2048,
timeout_seconds=60.0,
)
result = dispatcher._run_local_dispatch(
entry,
_messages("system", "user"),
category=None,
body={},
session_key=session_key,
)
assert result["payload"]["choices"][0]["message"]["content"] == "local answer"
assert result["request_id"].startswith("local-dispatch-")
rows = _rows(db_path, "local_energy_observations")
assert rows, "expected a local_energy_observations row"
return rows, client
def test_local_dispatch_answer_records_conversation_key(
tmp_path, monkeypatch
):
rows, _ = _local_dispatch_harness(tmp_path, monkeypatch, session_key="c:ener")
assert rows[-1]["session_key"] == "c:ener"
# --- Acceptance 7 [H5]: /dispatch (no header) keeps local energy NULL -------
def test_no_header_local_dispatch_keeps_energy_null_and_resolves(
tmp_path, monkeypatch
):
# /dispatch passes no session_key to _run_local_dispatch, so the keyword
# defaults to None and the local energy row must stay NULL.
rows, _ = _local_dispatch_harness(tmp_path, monkeypatch, session_key=None)
assert rows[-1]["session_key"] is None
# --- Acceptance 8 [H4]: cold start — sensible routing on a short history ----
def test_cold_start_conversation_routes_sensibly_on_short_history(
tmp_path, monkeypatch
):
client, db_path = make_router(tmp_path, monkeypatch)
# A short sub-agent history (3 messages) with a fresh conversation key
# that has NO prior route_decisions rows: must still resolve to a real
# model instead of failing or guessing an incumbent that cannot exist.
msgs = _messages("system", "user", "assistant")
r = _post(
client,
headers={"X-Router-Conversation": "cold"},
msgs=msgs,
)
assert r.status_code == 200, r.text
row = _latest_decision(db_path)
assert row["kind"] == "chat"
assert row["session_key"] == "c:cold"
assert row["selected_model"] is not None

39
tests/test_docs_content.py Normal file
View File

@@ -0,0 +1,39 @@
"""Content checks on the shipped documentation prose.
These tests read the actual docs files from the repo and assert on their
prose, so a doc edit that drifts from the contract is caught the same way a
code regression is. All offline, no network.
"""
from pathlib import Path
import pytest
DOCS = Path(__file__).resolve().parent.parent / "docs"
@pytest.fixture(scope="module")
def api_doc() -> str:
doc = DOCS / "api.md"
assert doc.exists(), f"docs/api.md not found at {doc}"
return doc.read_text(encoding="utf-8")
def test_api_doc_documents_conversation_header_name(api_doc: str) -> None:
"""The conversation header must be named in the API reference."""
assert "X-Router-Conversation" in api_doc
assert "X-Router-Agent" in api_doc
assert "X-Router-Parent" in api_doc
def test_api_doc_documents_conversation_id_charset_rule(api_doc: str) -> None:
"""The charset and length contract for the conversation id must appear."""
assert "1-128" in api_doc
assert "^[A-Za-z0-9._:-]+$" in api_doc
def test_api_doc_documents_fallback_behavior(api_doc: str) -> None:
"""Absent or invalid headers must be documented as falling back to absent,
and an absent conversation header as falling back to the fingerprint."""
assert "treated as absent" in api_doc
assert "fingerprint" in api_doc

224
tests/test_metrics.py
View File

@@ -36,6 +36,7 @@ from metrics import (
capability_ceilings,
capability_demand_warnings,
context_ceilings,
conversation_adoption,
cumulative_spend_series,
cumulative_spend_warnings,
demand_ceiling_warnings,
@@ -2556,3 +2557,226 @@ def test_scoring_coverage_carries_the_premise_expiry_series(tmp_path):
w.startswith("cost-as-tiebreak premise expired")
for w in coverage["warnings"]
)
# --- conversation adoption tests ----------------------------------------------
def _seed_conversation_adoption_rows(
conn: sqlite3.Connection,
*,
c_rows: tuple = (),
fingerprint_in_window: tuple = (),
) -> None:
"""Seed route_decisions with c:-prefixed and fingerprint session_keys.
Each c-row is a (session_key, in_window) pair; a fingerprint row needs
only its in-window flag.
"""
now = _now()
for key, in_window in c_rows:
conn.execute(
"INSERT INTO route_decisions "
"(observed_at, kind, session_key, agent, parent_key) "
"VALUES (?, 'chat', ?, 'classifier', NULL)",
(
(now - timedelta(hours=1)).isoformat() if in_window
else (now - timedelta(days=30)).isoformat(),
key,
),
)
for i, in_window in enumerate(fingerprint_in_window):
conn.execute(
"INSERT INTO route_decisions "
"(observed_at, kind, session_key, agent, parent_key) "
"VALUES (?, 'chat', ?, 'classifier', NULL)",
(
(now - timedelta(hours=1)).isoformat() if in_window
else (now - timedelta(days=30)).isoformat(),
f"fp-{i}",
),
)
conn.commit()
def test_conversation_adoption_counts_conversations_and_decisions(tmp_path):
"""3 c: rows over 2 keys + 1 fingerprint row -> 2 conversations,
3 c: decisions, 4 decisions, share 0.75."""
conn = _make_db(tmp_path)
_seed_conversation_adoption_rows(
conn,
c_rows=(("c:conv-a", True), ("c:conv-a", True), ("c:conv-b", True)),
fingerprint_in_window=(True,),
)
result = conversation_adoption(conn, CFG)
conn.close()
assert result["n_c_conversations"] == 2
assert result["n_c_decisions"] == 3
assert result["n_decisions"] == 4
assert result["share"] == pytest.approx(0.75)
assert result["window_seconds"] == 604800 # from config.yaml
def test_conversation_adoption_empty_table(tmp_path):
"""An empty route_decisions -> 0, 0, 0 and share None."""
conn = _make_db(tmp_path)
result = conversation_adoption(conn, CFG)
conn.close()
assert result["n_c_conversations"] == 0
assert result["n_c_decisions"] == 0
assert result["n_decisions"] == 0
assert result["share"] is None
def test_conversation_adoption_window_excludes_old_rows(tmp_path):
"""The window applies to every count: a 30 d old c: row and fingerprint
row stay out of n_c_* AND the n_decisions denominator."""
conn = _make_db(tmp_path)
_seed_conversation_adoption_rows(
conn,
c_rows=(
("c:conv-a", True),
("c:conv-b", True),
("c:conv-old", False),
),
fingerprint_in_window=(False,),
)
result = conversation_adoption(conn, CFG)
conn.close()
assert result["n_c_conversations"] == 2
assert result["n_c_decisions"] == 2
assert result["n_decisions"] == 2
def test_conversation_adoption_db_without_session_key_column(tmp_path):
"""DB whose route_decisions predates the identity migration -> the gate
dict: zeros, share None, no error."""
conn = sqlite3.connect(str(tmp_path / "no_session_key.db"))
conn.row_factory = sqlite3.Row
conn.execute(
"CREATE TABLE route_decisions "
"(id INTEGER PRIMARY KEY, observed_at TEXT, kind TEXT)"
)
conn.commit()
result = conversation_adoption(conn, CFG)
conn.close()
assert result["n_c_conversations"] == 0
assert result["n_c_decisions"] == 0
assert result["n_decisions"] == 0
assert result["share"] is None
assert result["window_seconds"] == 604800 # from config.yaml
# --- [H3] migration-path tests ------------------------------------------------
def test_ensure_tables_adds_route_decisions_columns_to_old_db(tmp_path, monkeypatch):
"""A DB created from the OLD schema gains both agent and parent_key columns."""
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
old_route_decisions = """
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,
flex_preference TEXT,
flex_swapped INTEGER,
flex_forced INTEGER,
request_id TEXT,
exploration INTEGER DEFAULT 0,
pinch_original_tokens INTEGER,
pinch_final_tokens INTEGER,
profile TEXT,
prefix_divergence_index INTEGER,
prefix_tokens_after_divergence INTEGER,
prefix_prev_message_count INTEGER
)
"""
db_path = tmp_path / "old-schema.db"
old_conn = sqlite3.connect(str(db_path))
old_conn.executescript(old_route_decisions)
old_conn.execute(
"CREATE TABLE local_energy_observations ("
"id INTEGER PRIMARY KEY, model_id TEXT, call_type TEXT, request_id TEXT,"
"session_dir TEXT, avg_power_watts REAL, duration_seconds REAL,"
"energy_kwh REAL, cost_usd REAL, carbon_g_co2eq REAL, meter TEXT,"
"observed_at TEXT)"
)
old_conn.execute(
"CREATE TABLE energy_observations (id INTEGER PRIMARY KEY, model_id TEXT, provider TEXT)"
)
old_conn.close()
# Point dispatcher at this DB and run _ensure_tables
import dispatcher as d
original_path = d.cfg.database.path
try:
d.cfg.database.path = str(db_path)
d._ensure_tables()
finally:
d.cfg.database.path = original_path
# Verify both columns exist
check = sqlite3.connect(str(db_path))
check.row_factory = sqlite3.Row
cols = {row[1] for row in check.execute("PRAGMA table_info(route_decisions)")}
assert "agent" in cols, "agent column should exist after migration"
assert "parent_key" in cols, "parent_key column should exist after migration"
# Verify local_energy_observations has session_key
local_cols = {row[1] for row in check.execute("PRAGMA table_info(local_energy_observations)")}
assert "session_key" in local_cols, "session_key column should exist after migration"
# Also test that recent_decisions runs cleanly on the migrated DB
rows = recent_decisions(check)
check.close()
assert isinstance(rows, list)
def test_log_local_energy_persists_session_key(tmp_path):
"""log_local_energy stores the session_key value given."""
conn = sqlite3.connect(str(tmp_path / "le-test.db"))
conn.row_factory = sqlite3.Row
import local_energy
local_energy.ensure_local_energy_table(conn)
local_energy.log_local_energy(
conn,
model_id="test-model",
call_type="classify",
avg_power_watts=50.0,
duration_seconds=10.0,
energy_kwh=0.00014,
cost_usd=0.0,
carbon_g_co2eq=0.0,
meter="test",
observed_at=_now().isoformat(),
session_key="c:conv-42",
)
row = conn.execute(
"SELECT session_key FROM local_energy_observations LIMIT 1"
).fetchone()
conn.close()
assert row is not None
assert row["session_key"] == "c:conv-42"

190
tests/test_no_header_snapshot.py Normal file
View File

@@ -0,0 +1,190 @@
"""Baseline no-header route_decisions snapshot (Task 3, pre-wiring fixture).
A committed test cannot read a gitignored ``.omo/evidence/`` path on a clean
checkout, so the pre-wiring ``route_decisions`` row for a HEADERLESS request
is frozen as an ordinary JSON fixture under ``tests/fixtures/`` and committed
as its OWN commit BEFORE the wiring that adds conversation identity.
The Success Criteria [H6] for identity wiring requires this snapshot to
reproduce byte-identically after wiring: a no-header request must still land
under the same session fingerprint with NULL agent/parent_key, so nothing
about the row may change. Columns that record measured wall-clock time are
excluded (``classifier_ms``, ``router_wall_seconds``, ``router_ttft_seconds``,
``observed_at``); ``id`` is excluded as it is an autoincrement.
This module owns the shared dump/compare helpers so ``test_conversation_wiring.py``
can reuse the exact same normalization when it asserts the byte-identical
snapshot as acceptance case 3.
"""
import json
import sqlite3
from pathlib import Path
import pytest
from starlette.testclient import TestClient
import dispatcher
from config import RoutingProfile
from dispatcher import Classification, app
FIXTURES_DIR = Path(__file__).parent / "fixtures"
SNAPSHOT_PATH = FIXTURES_DIR / "route_decision_no_header.json"
# Columns that record measured wall-clock time (or an autoincrement id) must
# never appear in the snapshot, or the frozen row would drift run to run.
# [H6] requires router_wall_seconds and router_ttft_seconds in this list.
EXCLUDED_COLS = (
"id",
"observed_at",
"classifier_ms",
"router_wall_seconds",
"router_ttft_seconds",
)
CHEAP = "cheap-model"
DEAR = "dear-model"
class FakeResponse:
"""Just enough of requests.Response for the dispatcher path."""
def __init__(self, payload=None, *, status_code=200):
self.status_code = status_code
self._payload = payload or {}
self.text = json.dumps(self._payload)
self.headers = {}
self.request = None
def json(self):
return self._payload
def _completion(model):
return {
"id": "chatcmpl-test",
"object": "chat.completion",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "ok"},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 30, "completion_tokens": 5},
"model": model,
}
def make_router(tmp_path, monkeypatch):
"""Point the dispatcher at a throwaway catalog with nothing dialled out.
Mirrors the ``router`` fixture in ``test_chat_completions.py``: two models
with a classifying stub and a fake requests.post, so one routed POST is
deterministic and offline. Returns ``(TestClient, db_path)``.
"""
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
schema = (Path(__file__).resolve().parent.parent / "config" / "schema.sql").read_text()
conn.executescript(schema)
for model_id, completion_price, vision in (
(CHEAP, 0.30, 1),
(DEAR, 9.00, 0),
):
conn.execute(
"""
INSERT INTO models (
model_id, provider, base_model_id, tier, context_window,
effective_context_window, max_output_tokens,
cost_per_1m_prompt, cost_per_1m_completion,
supports_vision, supports_json_mode,
latency_class, reasoning_mode, context_variant,
access_level, availability, last_updated
) VALUES (?, 'neuralwatt', ?, 2, 262128, 192500, 16384, ?, ?,
?, 1, 'standard', 'default', 'full', 'public', 'active',
'2026-08-22T00:00:00+00:00')
""",
(model_id, model_id, completion_price / 3, completion_price, vision),
)
conn.commit()
conn.close()
monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", False)
monkeypatch.setattr(dispatcher.cfg.session_cache, "enabled", False)
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", False)
monkeypatch.setattr(dispatcher.cfg.exploration, "epsilon", 0.0)
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
def fake_post(url, headers=None, json=None, stream=False, timeout=None):
return FakeResponse(_completion(json["model"]))
monkeypatch.setattr(dispatcher.requests, "post", fake_post)
monkeypatch.setattr(
dispatcher, "classify",
lambda task, context: Classification(
task_category="coding_general", task_tier=2,
required_context_tokens=100, confidence=0.9,
),
)
return TestClient(app), db_path
def no_header_payload():
return {"model": "auto", "messages": [{"role": "user", "content": "write a function"}]}
def post_no_header(client):
"""POST one headerless routed request, returning the raw response."""
return client.post("/v1/chat/completions", json=no_header_payload())
def normalized_route_decision(db_path):
"""The single route_decisions row as a normalized dict for snapshotting.
Returns the row with ``id``/``observed_at``/``classifier_ms``/
``router_wall_seconds``/``router_ttft_seconds`` dropped and None-values
kept (so agent/parent_key NULL is visible and byte-stable).
"""
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM route_decisions ORDER BY id DESC LIMIT 1"
).fetchone()
conn.close()
assert row is not None, "expected exactly one route_decisions row"
return {
key: row[key]
for key in row.keys()
if key not in EXCLUDED_COLS and not key.startswith("__")
}
def serialize(data):
"""Byte-stable JSON serialization for the snapshot comparison."""
return json.dumps(data, sort_keys=True, indent=2, ensure_ascii=True) + "\n"
def assert_matches_snapshot(db_path):
"""Assert the current no-header row reproduces the committed snapshot."""
current = normalized_route_decision(db_path)
assert SNAPSHOT_PATH.exists(), (
f"snapshot fixture missing; run the generator to write {SNAPSHOT_PATH}"
)
expected = json.loads(SNAPSHOT_PATH.read_text())
assert current == expected, (
"no-header route_decisions row drifted from committed snapshot:\n"
f"current={serialize(current)}\n"
f"expected={serialize(expected)}"
)
# Byte-identical: the serialized forms must match exactly.
assert serialize(current) == SNAPSHOT_PATH.read_text()
def test_no_header_route_decision_matches_committed_snapshot(tmp_path, monkeypatch):
"""A headerless request reproduces the committed pre-wiring snapshot."""
client, db_path = make_router(tmp_path, monkeypatch)
resp = post_no_header(client)
assert resp.status_code == 200, resp.text
assert_matches_snapshot(db_path)

212
tests/test_outcome_attribution.py
View File

@@ -22,9 +22,15 @@ from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
from starlette.testclient import TestClient
import dispatcher
from dispatcher import AMBIGUOUS, SEED_CATEGORY, _most_recent_if_unambiguous
from dispatcher import (
AMBIGUOUS,
SEED_CATEGORY,
_find_outcome_row,
_most_recent_if_unambiguous,
)
from metrics import quota_burn
ROOT = Path(__file__).resolve().parent.parent
@@ -63,15 +69,16 @@ def db(tmp_path):
def _observe(conn, *, when: datetime, session_key: str, request_id: str,
category: str = "coding_general"):
category: str = "coding_general", session_dir: object = None):
conn.execute(
"""
INSERT INTO energy_observations (
model_id, provider, request_id, session_key, task_category, observed_at
) VALUES ('m', 'neuralwatt', ?, ?, ?, ?)
model_id, provider, request_id, session_key, task_category,
session_dir, observed_at
) VALUES ('m', 'neuralwatt', ?, ?, ?, ?, ?)
""",
# Written exactly the way log_observation writes it.
(request_id, session_key, category, when.isoformat()),
(request_id, session_key, category, session_dir, when.isoformat()),
)
conn.commit()
@@ -226,3 +233,198 @@ def test_degraded_sources_match_the_cascade_steps():
)
trusted = {"classifier", "override", "cached", "classifier_cloud"}
assert dispatcher.DEGRADED_CLASSIFICATION_SOURCES == literal_sources - trusted
# ---------------------------------------------------------------------------
# Conversation-id attribution (Contract rule 4)
# ---------------------------------------------------------------------------
def _conversation_key(conversation_id: str) -> str:
return "c:" + conversation_id
def _observe_local(conn, *, when: datetime, request_id: str,
session_key: object = None, session_dir: object = None,
call_type: str = "coding_general"):
conn.execute(
"""
INSERT INTO local_energy_observations (
model_id, call_type, request_id, session_dir, session_key, observed_at
) VALUES ('m', ?, ?, ?, ?, ?)
""",
(call_type, request_id, session_dir, session_key, when.isoformat()),
)
conn.commit()
def _open_db(path) -> sqlite3.Connection:
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
return conn
def test_conversation_id_resolves_exact_conversation_over_shared_source_dir(db):
"""Two conversations under one cwd: naming one picks it exactly.
Both rows sit inside the window and share a session_dir, so the old path
(no request_id, no conversation_id) can only refuse -- two live
conversations is exactly the ambiguous case. conversation_id cuts through.
"""
now = _now()
_observe(db, when=now, session_key=_conversation_key("older"),
request_id="old", session_dir="/shared")
_observe(db, when=now, session_key=_conversation_key("newer"),
request_id="new", session_dir="/shared")
# Both live: the fallback can't pick a winner -- the 409 case.
assert _most_recent_if_unambiguous(db) is AMBIGUOUS
row = _find_outcome_row(db, None, None, conversation_id="older")
assert row is not None and row is not AMBIGUOUS
assert row["request_id"] == "old"
def test_conversation_id_resolves_even_when_request_id_omitted(db):
"""A lone conversation is picked by conversation_id alone (no request_id)."""
now = _now()
_observe(db, when=now, session_key=_conversation_key("conv"), request_id="rid")
row = _find_outcome_row(db, None, None, conversation_id="conv")
assert row is not None and row is not AMBIGUOUS
assert row["request_id"] == "rid"
def test_request_id_wins_over_conversation_id(db):
"""Contract rule 4: request_id resolves first, even when it disagrees."""
now = _now()
_observe(db, when=now, session_key=_conversation_key("by-conv"), request_id="want")
_observe(db, when=now, session_key=_conversation_key("other"), request_id="other")
row = _find_outcome_row(db, "want", None, conversation_id="other")
assert row["request_id"] == "want"
def test_local_dispatch_row_attributed_by_conversation_id(db):
"""A local-dispatch answer reports in by conversation_id too."""
now = _now()
_observe_local(db, when=now, session_key=_conversation_key("conv"),
request_id="local-rid")
row = _find_outcome_row(db, None, None, conversation_id="conv")
assert row is not None and row is not AMBIGUOUS
assert row["provider"] == "ollama-local"
assert row["request_id"] == "local-rid"
def test_no_header_local_row_resolves_through_session_dir(db):
"""A local row with NULL session_key (no header) still resolves.
COALESCE(session_key, session_dir) keeps pre-header local rows joinable by
their working directory, exactly as the old session_dir-only expression did.
"""
now = _now()
_observe_local(db, when=now, session_key=None, session_dir="/work",
request_id="rid")
row = _most_recent_if_unambiguous(db)
assert row is not None and row is not AMBIGUOUS
assert row["request_id"] == "rid"
@pytest.fixture
def client(tmp_path, monkeypatch):
"""A TestClient wired to a temp DB seeded from schema.sql."""
conn = _open_db(tmp_path / "test.db")
conn.executescript(SCHEMA_SQL)
conn.close()
monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
with TestClient(dispatcher.app) as c:
yield c
def test_unknown_conversation_id_404_and_writes_nothing(client, tmp_path):
"""An unmatched conversation_id is a 404, and records no outcome."""
resp = client.post("/outcome", json={"ok": True, "conversation_id": "no-such"})
assert resp.status_code == 404
with _open_db(tmp_path / "test.db") as conn:
assert conn.execute("SELECT COUNT(*) FROM verifications").fetchone()[0] == 0
def test_invalid_conversation_id_is_422(client):
"""A present-but-invalid conversation_id is a client bug: 422."""
resp = client.post(
"/outcome", json={"ok": True, "conversation_id": "has a space"}
)
assert resp.status_code == 422
def test_h3_local_table_predates_session_key_migrates_on_first_request(
tmp_path, monkeypatch
):
"""A DB whose local_energy_observations lacks session_key still works.
SQLite cannot drop a column, so the table is recreated WITHOUT session_key
to simulate one that predates the column, then _ensure_tables migrates it.
Outcome attribution must not 500 against the pre-migration schema.
"""
db_path = tmp_path / "test.db"
conn = _open_db(db_path)
conn.executescript(SCHEMA_SQL)
conn.execute("DROP TABLE local_energy_observations")
conn.execute(
"""
CREATE TABLE local_energy_observations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
model_id TEXT NOT NULL,
call_type TEXT NOT NULL,
request_id TEXT,
session_dir TEXT,
avg_power_watts REAL,
duration_seconds REAL,
energy_kwh REAL,
cost_usd REAL,
carbon_g_co2eq REAL,
meter TEXT,
observed_at TEXT NOT NULL
)
"""
)
conn.execute(
"""
INSERT INTO energy_observations (
model_id, provider, request_id, session_key, task_category, observed_at
) VALUES ('m', 'neuralwatt', 'cloud-rid', 'c:conv', 'coding_general', ?)
""",
(_now().isoformat(),),
)
conn.commit()
conn.close()
# The import-time _ensure_tables already ran against the real path; point
# it at OUR temp DB so the migration covers this pre-column table.
monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
dispatcher._ensure_tables()
# Idempotent: running the migration again must be a no-op.
dispatcher._ensure_tables()
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
with TestClient(dispatcher.app) as client:
seeded = client.post(
"/outcome", json={"ok": True, "request_id": "cloud-rid"}
)
assert seeded.status_code == 200
unknown = client.post(
"/outcome", json={"ok": True, "conversation_id": "unknown"}
)
assert unknown.status_code == 404

2
tests/test_route_decisions.py
View File

@@ -70,6 +70,8 @@ ROUTE_DECISIONS_COLUMNS = [
"prefix_divergence_index",
"prefix_tokens_after_divergence",
"prefix_prev_message_count",
"agent",
"parent_key",
]

14
tests/test_tui_schema_drift.py
View File

@@ -13,7 +13,7 @@ Two registries, not a magic diff:
``task_tier`` -> ``tier``, ``selected_model`` -> ``selected``), which is
exactly why the mapping is explicit and never derived by string identity.
- ``UNSURFACED_COLUMNS`` — columns deliberately NOT surfaced, each with a
one-line reason. Empty today.
one-line reason. Currently holds one entry (``parent_key``).
Every new schema column must land in one of the two registries or
``test_schema_columns_all_have_surfacing_decisions`` fails naming the column;
@@ -74,12 +74,14 @@ SCHEMA_TO_MODEL_KEYS = {
"prefix_divergence_index": "prefix_divergence_index",
"prefix_tokens_after_divergence": "prefix_tokens_after_divergence",
"prefix_prev_message_count": "prefix_prev_message_count",
"agent": "agent",
}
# Columns deliberately NOT surfaced anywhere in the TUI get recorded here with a
# one-line reason (callers must keep the comment). Empty today: every column is
# surfaced via decision_row. A new schema column that lands in NEITHER registry
# fails the drift test.
UNSURFACED_COLUMNS: set[str] = set()
# one-line reason (callers must keep the comment). A new schema column that lands
# in NEITHER registry fails the drift test.
UNSURFACED_COLUMNS: set[str] = {
"parent_key", # opaque id, recorded for later lineage queries, not displayed
}
# Every column seeded non-NULL: the single source of truth for the INSERT and
# the round-trip assertions, so the tests prove the column flows, not that
@@ -122,6 +124,8 @@ FULL_ROW = {
"prefix_divergence_index": 19,
"prefix_tokens_after_divergence": 72091,
"prefix_prev_message_count": 81,
"agent": "classifier",
"parent_key": None,
}