feat: per-model throughput (tok/s) tracking + metrics endpoint

New throughput-store extension records one token-weighted sample per turn
(model, output tokens, pure generation time = Σ step genTotalMs) into a
day-bucketed KV store, and aggregates per-model tok/s = Σtokens / Σgen-seconds
over a day/week/month (server-local boundaries; week = ISO Mon–Sun).

transport-http records a sample per turn (logged) and serves
GET /metrics/throughput?period=day|week|month&date=<...>. The response is typed
as transport-contract's ThroughputResponse, so store/wire drift is a compile
error. Pure period + aggregate logic fully unit-tested.

4 files changed, 220 insertions, 4 deletions

diff --git a/packages/transport-http/src/app.test.ts b/packages/transport-http/src/app.test.ts
index 0a6c5b0..e634e19 100644
--- a/packages/transport-http/src/app.test.ts
+++ b/packages/transport-http/src/app.test.ts
@@ -1,8 +1,33 @@
-import type { AgentEvent, Logger, StepId, StoredChunk, TurnMetrics } from "@dispatch/kernel";
+import type {
+	AgentEvent,
+	Logger,
+	StepId,
+	StorageNamespace,
+	StoredChunk,
+	TurnMetrics,
+} from "@dispatch/kernel";
+import { createThroughputStore, dayKeyOf } from "@dispatch/throughput-store";
+import type { ThroughputResponse } from "@dispatch/transport-contract";
 import { describe, expect, it } from "vitest";
 import { createApp } from "./app.js";
 import type { ConversationStore, CredentialStore, SessionOrchestrator } from "./seam.js";
 
+function createMemStorage(): StorageNamespace {
+	const map = new Map<string, string>();
+	return {
+		get: async (k) => map.get(k) ?? null,
+		set: async (k, v) => {
+			map.set(k, v);
+		},
+		delete: async (k) => {
+			map.delete(k);
+		},
+		has: async (k) => map.has(k),
+		keys: async (prefix) =>
+			[...map.keys()].filter((k) => (prefix === undefined ? true : k.startsWith(prefix))),
+	};
+}
+
 interface CapturedLog {
 	readonly level: "debug" | "info" | "warn" | "error";
 	readonly msg: string;
@@ -739,3 +764,114 @@ describe("CORS", () => {
 		expect(res.headers.get("Access-Control-Allow-Headers")).toContain("Content-Type");
 	});
 });
+
+describe("throughput recording + GET /metrics/throughput", () => {
+	const ts = new Date(2026, 5, 10, 12, 0, 0).getTime();
+	const day = dayKeyOf(ts);
+
+	function appWith(
+		throughputStore: ReturnType<typeof createThroughputStore>,
+		events: AgentEvent[],
+	) {
+		return createApp({
+			conversationStore: createFakeConversationStore(),
+			orchestrator: createFakeOrchestrator(events),
+			credentialStore: createFakeCredentialStore([]),
+			throughputStore,
+			now: () => ts,
+		});
+	}
+
+	async function postChat(app: ReturnType<typeof createApp>, body: Record<string, unknown>) {
+		return app.request("/chat", {
+			method: "POST",
+			headers: { "Content-Type": "application/json" },
+			body: JSON.stringify(body),
+		});
+	}
+
+	it("records a per-model sample from a turn and aggregates it (token-weighted tok/s)", async () => {
+		const store = createThroughputStore({ storage: createMemStorage() });
+		const events: AgentEvent[] = [
+			{
+				type: "step-complete",
+				conversationId: "c1",
+				turnId: "t1",
+				stepId: "t1#0" as StepId,
+				genTotalMs: 2000,
+			},
+			{
+				type: "done",
+				conversationId: "c1",
+				turnId: "t1",
+				reason: "stop",
+				usage: { inputTokens: 10, outputTokens: 400 },
+			},
+		];
+		const app = appWith(store, events);
+
+		const chat = await postChat(app, {
+			conversationId: "c1",
+			message: "hi",
+			model: "claude/haiku",
+		});
+		expect(chat.status).toBe(200);
+
+		const res = await app.request(`/metrics/throughput?period=day&date=${day}`);
+		expect(res.status).toBe(200);
+		const report = (await res.json()) as ThroughputResponse;
+		expect(report.period).toBe("day");
+		expect(report.models).toHaveLength(1);
+		expect(report.models[0]).toMatchObject({
+			model: "claude/haiku",
+			totalOutputTokens: 400,
+			totalGenMs: 2000,
+			tokensPerSecond: 200, // 400 tokens / 2s
+			turns: 1,
+		});
+	});
+
+	it("does not record a sample when no model is selected", async () => {
+		const store = createThroughputStore({ storage: createMemStorage() });
+		const events: AgentEvent[] = [
+			{
+				type: "step-complete",
+				conversationId: "c1",
+				turnId: "t1",
+				stepId: "t1#0" as StepId,
+				genTotalMs: 2000,
+			},
+			{
+				type: "done",
+				conversationId: "c1",
+				turnId: "t1",
+				reason: "stop",
+				usage: { inputTokens: 1, outputTokens: 5 },
+			},
+		];
+		const app = appWith(store, events);
+
+		await postChat(app, { conversationId: "c1", message: "hi" }); // no model
+		const res = await app.request(`/metrics/throughput?period=day&date=${day}`);
+		const report = (await res.json()) as { models: unknown[] };
+		expect(report.models).toEqual([]);
+	});
+
+	it("returns 400 for an invalid period", async () => {
+		const app = appWith(createThroughputStore({ storage: createMemStorage() }), []);
+		const res = await app.request("/metrics/throughput?period=year&date=2026");
+		expect(res.status).toBe(400);
+	});
+
+	it("returns 400 for a malformed date", async () => {
+		const app = appWith(createThroughputStore({ storage: createMemStorage() }), []);
+		const res = await app.request("/metrics/throughput?period=day&date=nope");
+		expect(res.status).toBe(400);
+	});
+
+	it("returns 400 when date is missing", async () => {
+		const app = appWith(createThroughputStore({ storage: createMemStorage() }), []);
+		const res = await app.request("/metrics/throughput?period=day");
+		expect(res.status).toBe(400);
+	});
+});
diff --git a/packages/transport-http/src/app.ts b/packages/transport-http/src/app.ts
index 4002e23..3c9ae85 100644
--- a/packages/transport-http/src/app.ts
+++ b/packages/transport-http/src/app.ts
@@ -3,6 +3,7 @@ import type {
 	ConversationHistoryResponse,
 	ConversationMetricsResponse,
 	ModelsResponse,
+	ThroughputResponse,
 } from "@dispatch/transport-contract";
 import { Hono } from "hono";
 import { cors } from "hono/cors";
@@ -13,14 +14,24 @@ import {
 	parseSinceSeq,
 	serializeEventLine,
 } from "./logic.js";
-import type { ConversationStore, CredentialStore, SessionOrchestrator } from "./seam.js";
+import {
+	type ConversationStore,
+	type CredentialStore,
+	type SessionOrchestrator,
+	ThroughputQueryError,
+	type ThroughputStore,
+} from "./seam.js";
 
 export interface CreateServerOptions {
 	readonly conversationStore: ConversationStore;
 	readonly orchestrator: SessionOrchestrator;
 	readonly credentialStore: CredentialStore;
+	/** Optional — defaults to a no-op store (recording disabled, empty reports). */
+	readonly throughputStore?: ThroughputStore;
 	readonly logger?: Logger;
 	readonly generateId?: () => string;
+	/** Injectable clock for sample timestamps (default Date.now). */
+	readonly now?: () => number;
 }
 
 const noopLogger: Logger = {
@@ -45,10 +56,44 @@ const noopLogger: Logger = {
 	},
 };
 
+const noopThroughputStore: ThroughputStore = {
+	record: async () => {},
+	aggregate: async (q) => ({ period: q.period, date: q.date, start: 0, end: 0, models: [] }),
+};
+
 export function createApp(opts: CreateServerOptions): Hono {
 	const app = new Hono();
 	const log = opts.logger ?? noopLogger;
 	const generateId = opts.generateId ?? (() => crypto.randomUUID());
+	const now = opts.now ?? (() => Date.now());
+	const throughputStore = opts.throughputStore ?? noopThroughputStore;
+
+	async function recordThroughput(
+		turnEvents: readonly AgentEvent[],
+		model: string | undefined,
+	): Promise<void> {
+		if (model === undefined) return; // no model selected → nothing to attribute
+		let genMs = 0;
+		let outputTokens = 0;
+		for (const e of turnEvents) {
+			if (e.type === "step-complete" && e.genTotalMs !== undefined) genMs += e.genTotalMs;
+			if (e.type === "done" && e.usage !== undefined) outputTokens = e.usage.outputTokens;
+		}
+		if (genMs <= 0) return; // no generation time → can't compute tok/s
+		try {
+			await throughputStore.record({ model, ts: now(), outputTokens, genMs });
+			log.info("throughput: turn recorded", {
+				model,
+				outputTokens,
+				genMs,
+				tokensPerSecond: Math.round((outputTokens / (genMs / 1000)) * 100) / 100,
+			});
+		} catch (err) {
+			log.warn("throughput: failed to record sample", {
+				error: err instanceof Error ? err.message : String(err),
+			});
+		}
+	}
 
 	app.use(
 		"*",
@@ -174,6 +219,11 @@ export function createApp(opts: CreateServerOptions): Hono {
 		await streamReady;
 		await orchestratorPromise.catch(() => {});
 
+		// Record a per-model throughput sample for this turn. Generation time is
+		// the PURE decode time — the sum of per-step genTotalMs (excludes tool
+		// waits) — and tokens are the turn's aggregate output tokens.
+		await recordThroughput(events, model);
+
 		const ndjson = events.map(serializeEventLine).join("");
 
 		return c.text(ndjson, 200, {
@@ -182,5 +232,28 @@ export function createApp(opts: CreateServerOptions): Hono {
 		});
 	});
 
+	app.get("/metrics/throughput", async (c) => {
+		const period = c.req.query("period");
+		const date = c.req.query("date");
+		if (period !== "day" && period !== "week" && period !== "month") {
+			return c.json({ error: "query param 'period' must be one of: day, week, month" }, 400);
+		}
+		if (date === undefined || date === "") {
+			return c.json({ error: "query param 'date' is required" }, 400);
+		}
+		try {
+			// Typed against the wire contract: if the store's report shape ever
+			// drifts from ThroughputResponse, this assignment fails to compile.
+			const body: ThroughputResponse = await throughputStore.aggregate({ period, date });
+			return c.json(body);
+		} catch (err) {
+			if (err instanceof ThroughputQueryError) {
+				return c.json({ error: err.message }, 400);
+			}
+			log.error("throughput: aggregate failed", { err });
+			return c.json({ error: "Failed to aggregate throughput" }, 502);
+		}
+	});
+
 	return app;
 }
diff --git a/packages/transport-http/src/extension.ts b/packages/transport-http/src/extension.ts
index adbf87f..4abd7aa 100644
--- a/packages/transport-http/src/extension.ts
+++ b/packages/transport-http/src/extension.ts
@@ -4,6 +4,7 @@ import {
 	conversationStoreHandle,
 	credentialStoreHandle,
 	sessionOrchestratorHandle,
+	throughputStoreHandle,
 } from "./seam.js";
 
 export const manifest: Manifest = {
@@ -12,9 +13,11 @@ export const manifest: Manifest = {
 	version: "0.0.0",
 	apiVersion: "^0.1.0",
 	trust: "bundled",
-	dependsOn: ["conversation-store", "credential-store", "session-orchestrator"],
+	dependsOn: ["conversation-store", "credential-store", "session-orchestrator", "throughput-store"],
 	capabilities: { network: true },
-	contributes: { routes: ["/chat", "/conversations/:id", "/health", "/models"] },
+	contributes: {
+		routes: ["/chat", "/conversations/:id", "/health", "/models", "/metrics/throughput"],
+	},
 	activation: "eager",
 };
 
@@ -32,12 +35,14 @@ export function createTransportHttpExtension(): Extension & {
 			const conversationStore = host.getService(conversationStoreHandle);
 			const orchestrator = host.getService(sessionOrchestratorHandle);
 			const credentialStore = host.getService(credentialStoreHandle);
+			const throughputStore = host.getService(throughputStoreHandle);
 			const logger = host.logger;
 
 			const app = createApp({
 				conversationStore,
 				orchestrator,
 				credentialStore,
+				throughputStore,
 				logger,
 			});
 
diff --git a/packages/transport-http/src/seam.ts b/packages/transport-http/src/seam.ts
index c7bfb74..7dbaa1b 100644
--- a/packages/transport-http/src/seam.ts
+++ b/packages/transport-http/src/seam.ts
@@ -4,3 +4,5 @@ export type { CredentialStore } from "@dispatch/credential-store";
 export { credentialStoreHandle } from "@dispatch/credential-store";
 export type { SessionOrchestrator } from "@dispatch/session-orchestrator";
 export { sessionOrchestratorHandle } from "@dispatch/session-orchestrator";
+export type { ThroughputStore } from "@dispatch/throughput-store";
+export { ThroughputQueryError, throughputStoreHandle } from "@dispatch/throughput-store";