summaryrefslogtreecommitdiffhomepage
path: root/packages/core
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-02 13:57:41 +0900
committerAdam Malczewski <[email protected]>2026-06-02 13:57:41 +0900
commitd27d97bb3aa0c13f4032bab54703ebb9e1c84c81 (patch)
treeb5bcfed65be5a4d27a7bbe6b46e338dcd489c2e0 /packages/core
parent3671b82cc624117476e30b95eaf7d2bc3b34ae28 (diff)
parentc1439ea8c677ddfd11c219de39c3e77c7e297a9b (diff)
downloaddispatch-d27d97bb3aa0c13f4032bab54703ebb9e1c84c81.tar.gz
dispatch-d27d97bb3aa0c13f4032bab54703ebb9e1c84c81.zip
Merge branch 'dev' into u3/agent-effort-level
# Conflicts: # packages/api/tests/agent-manager.test.ts
Diffstat (limited to 'packages/core')
-rw-r--r--packages/core/src/chunks/transform.ts5
-rw-r--r--packages/core/src/db/chunks.ts85
-rw-r--r--packages/core/src/index.ts7
-rw-r--r--packages/core/src/models/catalog.ts179
-rw-r--r--packages/core/src/models/index.ts4
-rw-r--r--packages/core/src/types/index.ts61
-rw-r--r--packages/core/tests/db/chunks.db.test.ts336
-rw-r--r--packages/core/tests/models/catalog.test.ts158
8 files changed, 824 insertions, 11 deletions
diff --git a/packages/core/src/chunks/transform.ts b/packages/core/src/chunks/transform.ts
index a4c6fc8..e8f4a18 100644
--- a/packages/core/src/chunks/transform.ts
+++ b/packages/core/src/chunks/transform.ts
@@ -209,6 +209,11 @@ export function groupRowsToMessages(rows: ChunkRow[]): MessageRow[] {
continue;
}
+ // Usage rows are an invisible side channel (persisted for the backend
+ // aggregate only). They're already query-excluded from getChunksForTab,
+ // so this is defensive insurance: never let one leak into render grouping.
+ if (row.type === "usage") continue;
+
// assistant / tool rows → part of the current assistant message
const c = ensureAssistant(row);
switch (row.type) {
diff --git a/packages/core/src/db/chunks.ts b/packages/core/src/db/chunks.ts
index 077259d..e0aadf3 100644
--- a/packages/core/src/db/chunks.ts
+++ b/packages/core/src/db/chunks.ts
@@ -5,7 +5,14 @@ import {
groupRowsToMessages,
type MessageRow,
} from "../chunks/transform.js";
-import type { ChunkData, ChunkRow, ChunkRowDraft, TextData } from "../types/index.js";
+import type {
+ ChunkData,
+ ChunkRow,
+ ChunkRowDraft,
+ TextData,
+ UsageData,
+ UsageStats,
+} from "../types/index.js";
import { getDatabase } from "./index.js";
// Re-export the DB-free transforms so existing barrel consumers
@@ -101,7 +108,7 @@ export function getChunksForTab(
const db = getDatabase();
if (!options) {
const rows = db
- .query("SELECT * FROM chunks WHERE tab_id = $tabId ORDER BY seq ASC")
+ .query("SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' ORDER BY seq ASC")
.all({ $tabId: tabId }) as Array<Record<string, unknown>>;
return rows.map(mapRow);
}
@@ -110,24 +117,28 @@ export function getChunksForTab(
if (limit !== undefined) {
const rows = db
.query(
- "SELECT * FROM chunks WHERE tab_id = $tabId AND seq < $before ORDER BY seq DESC LIMIT $limit",
+ "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' AND seq < $before ORDER BY seq DESC LIMIT $limit",
)
.all({ $tabId: tabId, $before: before, $limit: limit }) as Array<Record<string, unknown>>;
return rows.map(mapRow).reverse();
}
const rows = db
- .query("SELECT * FROM chunks WHERE tab_id = $tabId AND seq < $before ORDER BY seq DESC")
+ .query(
+ "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' AND seq < $before ORDER BY seq DESC",
+ )
.all({ $tabId: tabId, $before: before }) as Array<Record<string, unknown>>;
return rows.map(mapRow).reverse();
}
if (limit !== undefined) {
const rows = db
- .query("SELECT * FROM chunks WHERE tab_id = $tabId ORDER BY seq DESC LIMIT $limit")
+ .query(
+ "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' ORDER BY seq DESC LIMIT $limit",
+ )
.all({ $tabId: tabId, $limit: limit }) as Array<Record<string, unknown>>;
return rows.map(mapRow).reverse();
}
const rows = db
- .query("SELECT * FROM chunks WHERE tab_id = $tabId ORDER BY seq ASC")
+ .query("SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' ORDER BY seq ASC")
.all({ $tabId: tabId }) as Array<Record<string, unknown>>;
return rows.map(mapRow);
}
@@ -145,11 +156,71 @@ export function getMessagesForTab(tabId: string): MessageRow[] {
export function getTotalChunkCount(tabId: string): number {
const db = getDatabase();
const row = db
- .query("SELECT COUNT(*) as count FROM chunks WHERE tab_id = $tabId")
+ .query("SELECT COUNT(*) as count FROM chunks WHERE tab_id = $tabId AND type != 'usage'")
.get({ $tabId: tabId }) as { count: number } | null;
return row?.count ?? 0;
}
+/**
+ * Aggregate per-tab token/cache usage across ALL persisted `usage` chunk rows.
+ *
+ * Usage rows are written as an invisible side channel (one row per `usage`
+ * AgentEvent) and are query-excluded from `getChunksForTab`/`getTotalChunkCount`,
+ * so this aggregate is the read path. Because it sums server-side over every
+ * row, it stays complete even after the frontend evicts/pages out old turns
+ * (eviction is in-memory only). The return shape is structurally identical to
+ * the frontend `CacheStats`, so reload can seed it directly.
+ *
+ * - cumulative `inputTokens`/`outputTokens`/`cacheReadTokens`/`cacheWriteTokens`
+ * = SUM over all usage rows;
+ * - `requests` = COUNT of usage rows;
+ * - `last` = the highest-seq usage row's split (most recent request);
+ * - `null` when the tab has no usage rows.
+ *
+ * Sums in JS after selecting the rows (mirroring `mapRow`) to avoid relying on
+ * `json_extract` over the freeform `data_json`.
+ */
+export function getUsageStatsForTab(tabId: string): UsageStats | null {
+ const db = getDatabase();
+ const rows = db
+ .query("SELECT data_json FROM chunks WHERE tab_id = $tabId AND type = 'usage' ORDER BY seq ASC")
+ .all({ $tabId: tabId }) as Array<{ data_json: string }>;
+ if (rows.length === 0) return null;
+
+ let inputTokens = 0;
+ let outputTokens = 0;
+ let cacheReadTokens = 0;
+ let cacheWriteTokens = 0;
+ let last: UsageData | null = null;
+ for (const row of rows) {
+ let u: UsageData;
+ try {
+ u = JSON.parse(row.data_json) as UsageData;
+ } catch {
+ continue;
+ }
+ inputTokens += u.inputTokens ?? 0;
+ outputTokens += u.outputTokens ?? 0;
+ cacheReadTokens += u.cacheReadTokens ?? 0;
+ cacheWriteTokens += u.cacheWriteTokens ?? 0;
+ last = {
+ inputTokens: u.inputTokens ?? 0,
+ outputTokens: u.outputTokens ?? 0,
+ cacheReadTokens: u.cacheReadTokens ?? 0,
+ cacheWriteTokens: u.cacheWriteTokens ?? 0,
+ };
+ }
+
+ return {
+ inputTokens,
+ outputTokens,
+ cacheReadTokens,
+ cacheWriteTokens,
+ requests: rows.length,
+ last,
+ };
+}
+
export function clearChunksForTab(tabId: string): void {
const db = getDatabase();
db.query("DELETE FROM chunks WHERE tab_id = $tabId").run({ $tabId: tabId });
diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts
index 327b0a5..7818024 100644
--- a/packages/core/src/index.ts
+++ b/packages/core/src/index.ts
@@ -37,6 +37,7 @@ export {
getChunksForTab,
getMessagesForTab,
getTotalChunkCount,
+ getUsageStatsForTab,
groupRowsToMessages,
type MessageRow,
} from "./db/chunks.js";
@@ -67,7 +68,11 @@ export {
} from "./llm/debug-logger.js";
export { createProvider } from "./llm/provider.js";
// Models
-export { ModelRegistry } from "./models/index.js";
+export {
+ getModelsCatalog,
+ ModelRegistry,
+ resolveContextLimit,
+} from "./models/index.js";
// Notifications (ntfy.sh)
export * from "./notifications/index.js";
export * from "./permission/index.js";
diff --git a/packages/core/src/models/catalog.ts b/packages/core/src/models/catalog.ts
new file mode 100644
index 0000000..dea4647
--- /dev/null
+++ b/packages/core/src/models/catalog.ts
@@ -0,0 +1,179 @@
+import { mkdirSync, readFileSync, renameSync, statSync, writeFileSync } from "node:fs";
+import { dirname } from "node:path";
+
+/**
+ * models.dev-backed model catalog. Resolves a model's MAXIMUM context window
+ * (`limit.context`) dynamically from the public models.dev API, mirroring how
+ * opencode determines per-model context limits — no hardcoded table.
+ *
+ * The catalog is fetched once, cached on disk with a short TTL, and reused. On
+ * fetch failure we fall back to a stale-but-present cache so the lookup keeps
+ * working offline. Lookups never throw: an unknown/unreachable model resolves
+ * to `null`, which the UI renders as "max unknown".
+ */
+
+/** Shape of the slice of models.dev's `/api.json` we consume. */
+interface ModelsDevModel {
+ limit?: {
+ context?: number;
+ output?: number;
+ };
+}
+
+interface ModelsDevProvider {
+ id: string;
+ models: Record<string, ModelsDevModel | undefined>;
+}
+
+type ModelsDevCatalog = Record<string, ModelsDevProvider | undefined>;
+
+/** Where models.dev's API lives. Overridable for tests / private mirrors. */
+const MODELS_URL = process.env.DISPATCH_MODELS_URL || "https://models.dev";
+
+/** Disk cache path (reuses the repo's `/tmp/dispatch` convention). */
+const CACHE_PATH = "/tmp/dispatch/models-dev.json";
+
+/** How long a cached catalog stays fresh before we re-fetch. */
+const CACHE_TTL_MS = 5 * 60 * 1000;
+
+/** Network timeout for the catalog fetch. */
+const FETCH_TIMEOUT_MS = 10_000;
+
+/**
+ * After a failed fetch we memoize the fallback for this long before retrying,
+ * so a sustained outage doesn't make every lookup hang on a fresh timeout.
+ */
+const FETCH_PENALTY_MS = 60_000;
+
+/**
+ * Dispatch provider id → models.dev provider ids to search, in priority order.
+ * We only support Claude-backed providers (per product scope). `anthropic` and
+ * `opencode-anthropic` are both Claude; we try the first-party `anthropic`
+ * catalog first, then the `opencode` gateway catalog as a fallback.
+ */
+const PROVIDER_MAP: Record<string, string[]> = {
+ anthropic: ["anthropic", "opencode"],
+ "opencode-anthropic": ["anthropic", "opencode"],
+};
+
+/** In-process memoized catalog promise (one fetch/parse per TTL window). */
+let cached: { catalog: ModelsDevCatalog; fetchedAt: number } | null = null;
+let inflight: Promise<ModelsDevCatalog> | null = null;
+
+function readDiskCache(): { catalog: ModelsDevCatalog; mtimeMs: number } | null {
+ try {
+ const stat = statSync(CACHE_PATH);
+ const text = readFileSync(CACHE_PATH, "utf-8");
+ return { catalog: JSON.parse(text) as ModelsDevCatalog, mtimeMs: stat.mtimeMs };
+ } catch {
+ return null;
+ }
+}
+
+function writeDiskCache(text: string): void {
+ try {
+ mkdirSync(dirname(CACHE_PATH), { recursive: true });
+ // Write-then-rename so a concurrent reader never sees a half-written
+ // file (rename is atomic on the same filesystem). The temp name is
+ // process-scoped to avoid two writers clobbering each other's temp.
+ const tmp = `${CACHE_PATH}.${process.pid}.tmp`;
+ writeFileSync(tmp, text, "utf-8");
+ renameSync(tmp, CACHE_PATH);
+ } catch {
+ // Best-effort: a read-only /tmp shouldn't break lookups.
+ }
+}
+
+async function fetchCatalog(): Promise<ModelsDevCatalog> {
+ const controller = new AbortController();
+ const timer = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS);
+ try {
+ const res = await fetch(`${MODELS_URL}/api.json`, { signal: controller.signal });
+ if (!res.ok) throw new Error(`models.dev returned HTTP ${res.status}`);
+ const text = await res.text();
+ const catalog = JSON.parse(text) as ModelsDevCatalog;
+ writeDiskCache(text);
+ return catalog;
+ } finally {
+ clearTimeout(timer);
+ }
+}
+
+/**
+ * Load the models.dev catalog, preferring in-process memo, then a fresh disk
+ * cache, then a network fetch. On network failure, falls back to any stale
+ * disk cache; if nothing is available, returns an empty catalog.
+ */
+export async function getModelsCatalog(): Promise<ModelsDevCatalog> {
+ if (process.env.DISPATCH_DISABLE_MODELS_FETCH) {
+ const disk = readDiskCache();
+ return disk?.catalog ?? {};
+ }
+
+ const now = Date.now();
+ if (cached && now - cached.fetchedAt < CACHE_TTL_MS) return cached.catalog;
+
+ // Fresh disk cache satisfies the request without a network round-trip.
+ const disk = readDiskCache();
+ if (disk && now - disk.mtimeMs < CACHE_TTL_MS) {
+ // Inherit the file's mtime as `fetchedAt` so loading a disk cache into
+ // a fresh process doesn't reset its TTL (which would otherwise double
+ // the worst-case staleness across process boundaries).
+ cached = { catalog: disk.catalog, fetchedAt: disk.mtimeMs };
+ return disk.catalog;
+ }
+
+ if (!inflight) {
+ inflight = fetchCatalog()
+ .then((catalog) => {
+ cached = { catalog, fetchedAt: Date.now() };
+ return catalog;
+ })
+ .catch((err) => {
+ // Network failed — serve a stale cache if we have one.
+ console.warn(
+ `dispatch: failed to fetch models.dev catalog: ${err instanceof Error ? err.message : String(err)}`,
+ );
+ const fallback = disk?.catalog ?? ({} as ModelsDevCatalog);
+ // Memoize the fallback with a short "penalty" TTL so a sustained
+ // outage doesn't make every lookup hang on a fresh 10s timeout.
+ // `fetchedAt` is backdated so the memo expires after FETCH_PENALTY_MS.
+ cached = {
+ catalog: fallback,
+ fetchedAt: Date.now() - CACHE_TTL_MS + FETCH_PENALTY_MS,
+ };
+ return fallback;
+ })
+ .finally(() => {
+ inflight = null;
+ });
+ }
+ return inflight;
+}
+
+/**
+ * Resolve a model's maximum context window (in tokens) for the given Dispatch
+ * provider + model id. Returns `null` when the provider is unsupported, the
+ * model is unknown, or the catalog is unavailable — callers should render that
+ * as "max unknown" (no denominator / percentage).
+ */
+export async function resolveContextLimit(
+ provider: string,
+ modelId: string,
+): Promise<number | null> {
+ const candidates = PROVIDER_MAP[provider];
+ if (!candidates || !modelId) return null;
+
+ const catalog = await getModelsCatalog();
+ for (const providerId of candidates) {
+ const ctx = catalog[providerId]?.models?.[modelId]?.limit?.context;
+ if (typeof ctx === "number" && ctx > 0) return ctx;
+ }
+ return null;
+}
+
+/** Test-only: reset the in-process memo so a test can re-exercise loading. */
+export function __resetCatalogCacheForTests(): void {
+ cached = null;
+ inflight = null;
+}
diff --git a/packages/core/src/models/index.ts b/packages/core/src/models/index.ts
index cf59749..2fcd657 100644
--- a/packages/core/src/models/index.ts
+++ b/packages/core/src/models/index.ts
@@ -1 +1,5 @@
+export {
+ getModelsCatalog,
+ resolveContextLimit,
+} from "./catalog.js";
export { ModelRegistry } from "./registry.js";
diff --git a/packages/core/src/types/index.ts b/packages/core/src/types/index.ts
index bab17f1..30afbd9 100644
--- a/packages/core/src/types/index.ts
+++ b/packages/core/src/types/index.ts
@@ -90,7 +90,14 @@ export interface ChatMessage {
export type ChunkRole = "user" | "assistant" | "tool" | "system";
/** Discriminator for a persisted chunk row's payload. */
-export type ChunkType = "text" | "thinking" | "tool_call" | "tool_result" | "error" | "system";
+export type ChunkType =
+ | "text"
+ | "thinking"
+ | "tool_call"
+ | "tool_result"
+ | "error"
+ | "system"
+ | "usage";
export interface TextData {
text: string;
@@ -119,6 +126,45 @@ export interface SystemData {
kind: SystemChunkKind;
text: string;
}
+/**
+ * Per-request token usage persisted as a SIDE-CHANNEL chunk row (one row per
+ * `usage` AgentEvent, i.e. one per LLM round-trip). These rows are deliberately
+ * EXCLUDED from `getChunksForTab`/`getTotalChunkCount` so they never enter the
+ * render, pagination, eviction, or agent-history-rebuild paths — they exist
+ * only to feed the backend aggregate `getUsageStatsForTab`, which seeds the
+ * frontend's `cacheStats` on reload. `inputTokens` is the TOTAL prompt
+ * (cached + fresh); `cacheReadTokens`/`cacheWriteTokens` are Anthropic's
+ * prompt-cache split. Mirrors the `usage` AgentEvent payload.
+ */
+export interface UsageData {
+ inputTokens: number;
+ outputTokens: number;
+ cacheReadTokens: number;
+ cacheWriteTokens: number;
+}
+
+/**
+ * Aggregate per-tab usage telemetry: the cumulative sum across ALL persisted
+ * `usage` rows, the request count, and the most recent request's split. This is
+ * the server-side source of truth (complete regardless of frontend
+ * eviction/pagination) returned by `getUsageStatsForTab`. Structurally
+ * identical to the frontend `CacheStats` so it can seed it directly. `null` when
+ * the tab has no usage rows.
+ */
+export interface UsageStats {
+ inputTokens: number;
+ outputTokens: number;
+ cacheReadTokens: number;
+ cacheWriteTokens: number;
+ /** Number of LLM requests (usage rows) counted. */
+ requests: number;
+ last: {
+ inputTokens: number;
+ outputTokens: number;
+ cacheReadTokens: number;
+ cacheWriteTokens: number;
+ } | null;
+}
export type ChunkData =
| TextData
@@ -126,7 +172,8 @@ export type ChunkData =
| ToolCallData
| ToolResultData
| ErrorData
- | SystemData;
+ | SystemData
+ | UsageData;
/**
* A persisted chunk row — the append-only unit of conversation storage and
@@ -225,8 +272,16 @@ export type AgentEvent =
* fold its transient live representation into the sealed chunk log. Emitted
* after `status: idle`/`error` (which fire before the DB write). Display/sync
* only — not conversation content.
+ *
+ * Carries `usageStats`: the tab's authoritative usage aggregate read from the
+ * DB AFTER the turn's usage rows were written. The frontend REPLACES (not adds)
+ * its live `cacheStats` with this, reconciling the live accumulator to the
+ * persisted truth every turn. This self-heals the live overshoot that occurs
+ * when a rate-limited fallback attempt's usage is streamed live but then
+ * discarded server-side (never persisted). `null` ⇒ tab has no usage rows;
+ * absent ⇒ leave `cacheStats` untouched (back-compat).
*/
- | { type: "turn-sealed"; turnId: string }
+ | { type: "turn-sealed"; turnId: string; usageStats?: UsageStats | null }
| { type: "text-delta"; delta: string }
| { type: "reasoning-delta"; delta: string }
/**
diff --git a/packages/core/tests/db/chunks.db.test.ts b/packages/core/tests/db/chunks.db.test.ts
new file mode 100644
index 0000000..4f7d517
--- /dev/null
+++ b/packages/core/tests/db/chunks.db.test.ts
@@ -0,0 +1,336 @@
+import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import type { ChunkRowDraft, UsageData } from "../../src/types/index.js";
+
+/**
+ * Internal row shape — matches the production `chunks` table columns.
+ * Kept loose at the `query()` boundary to mirror bun:sqlite's dynamic
+ * return type.
+ */
+interface ChunkRecord {
+ id: string;
+ tab_id: string;
+ seq: number;
+ turn_id: string;
+ step: number;
+ role: string;
+ type: string;
+ data_json: string;
+ created_at: number;
+}
+
+/**
+ * In-memory fake of `bun:sqlite`'s Database implementing only the queries
+ * `chunks.ts` actually issues. Same approach as `tabs.test.ts`: match exact
+ * normalized query strings as fixed branches (no SQL parser), so a query-string
+ * change fails loudly as "unsupported" instead of silently returning wrong data.
+ *
+ * This lets the DB-backed `getChunksForTab` / `getTotalChunkCount` /
+ * `getUsageStatsForTab` logic run under vitest, where `bun:sqlite` can't load.
+ */
+class FakeDatabase {
+ rows: ChunkRecord[] = [];
+ private idCounter = 0;
+
+ query(sql: string): {
+ all: (params?: Record<string, unknown>) => unknown[];
+ get: (params?: Record<string, unknown>) => unknown;
+ run: (params?: Record<string, unknown>) => void;
+ } {
+ return {
+ all: (params) => this.execSelect(sql, params),
+ get: (params) => this.execSelect(sql, params)[0] ?? null,
+ run: (params) => {
+ this.execMutation(sql, params);
+ },
+ };
+ }
+
+ /** bun:sqlite's `db.transaction(fn)` returns a callable that runs `fn`. */
+ transaction(fn: () => void): () => void {
+ return () => {
+ fn();
+ };
+ }
+
+ private execSelect(sql: string, params?: Record<string, unknown>): unknown[] {
+ const norm = sql.replace(/\s+/g, " ").trim();
+ const tabId = params?.$tabId as string | undefined;
+ const forTab = this.rows.filter((r) => r.tab_id === tabId);
+ const visible = forTab.filter((r) => r.type !== "usage");
+
+ // appendChunks: next-seq lookup (counts ALL rows, incl. usage)
+ if (norm === "SELECT COALESCE(MAX(seq), -1) as max_seq FROM chunks WHERE tab_id = $tabId") {
+ const seqs = forTab.map((r) => r.seq);
+ return [{ max_seq: seqs.length > 0 ? Math.max(...seqs) : -1 }];
+ }
+
+ // getChunksForTab — no options (usage excluded)
+ if (
+ norm === "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' ORDER BY seq ASC"
+ ) {
+ return [...visible].sort((a, b) => a.seq - b.seq);
+ }
+
+ // getChunksForTab — before + limit (usage excluded)
+ if (
+ norm ===
+ "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' AND seq < $before ORDER BY seq DESC LIMIT $limit"
+ ) {
+ const before = params?.$before as number;
+ const limit = params?.$limit as number;
+ return visible
+ .filter((r) => r.seq < before)
+ .sort((a, b) => b.seq - a.seq)
+ .slice(0, limit);
+ }
+
+ // getChunksForTab — before only (usage excluded)
+ if (
+ norm ===
+ "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' AND seq < $before ORDER BY seq DESC"
+ ) {
+ const before = params?.$before as number;
+ return visible.filter((r) => r.seq < before).sort((a, b) => b.seq - a.seq);
+ }
+
+ // getChunksForTab — limit only (usage excluded)
+ if (
+ norm ===
+ "SELECT * FROM chunks WHERE tab_id = $tabId AND type != 'usage' ORDER BY seq DESC LIMIT $limit"
+ ) {
+ const limit = params?.$limit as number;
+ return [...visible].sort((a, b) => b.seq - a.seq).slice(0, limit);
+ }
+
+ // getTotalChunkCount (usage excluded)
+ if (norm === "SELECT COUNT(*) as count FROM chunks WHERE tab_id = $tabId AND type != 'usage'") {
+ return [{ count: visible.length }];
+ }
+
+ // getUsageStatsForTab: usage rows only, in seq order
+ if (
+ norm ===
+ "SELECT data_json FROM chunks WHERE tab_id = $tabId AND type = 'usage' ORDER BY seq ASC"
+ ) {
+ return forTab
+ .filter((r) => r.type === "usage")
+ .sort((a, b) => a.seq - b.seq)
+ .map((r) => ({ data_json: r.data_json }));
+ }
+
+ throw new Error(`FakeDatabase: unsupported SELECT: ${norm}`);
+ }
+
+ private execMutation(sql: string, params?: Record<string, unknown>): void {
+ const norm = sql.replace(/\s+/g, " ").trim();
+
+ // appendChunks: single-row insert
+ if (
+ norm ===
+ "INSERT INTO chunks (id, tab_id, seq, turn_id, step, role, type, data_json, created_at) VALUES ($id, $tabId, $seq, $turnId, $step, $role, $type, $dataJson, $now)"
+ ) {
+ this.rows.push({
+ id: (params?.$id as string) ?? `c${this.idCounter++}`,
+ tab_id: params?.$tabId as string,
+ seq: params?.$seq as number,
+ turn_id: params?.$turnId as string,
+ step: (params?.$step as number) ?? 0,
+ role: params?.$role as string,
+ type: params?.$type as string,
+ data_json: params?.$dataJson as string,
+ created_at: (params?.$now as number) ?? 0,
+ });
+ return;
+ }
+
+ throw new Error(`FakeDatabase: unsupported mutation: ${norm}`);
+ }
+}
+
+let fakeDb: FakeDatabase;
+
+vi.mock("../../src/db/index.js", () => ({
+ getDatabase: vi.fn(() => fakeDb),
+}));
+
+const { appendChunks, getChunksForTab, getTotalChunkCount, getUsageStatsForTab } = await import(
+ "../../src/db/chunks.js"
+);
+
+function usageDraft(turnId: string, u: UsageData): ChunkRowDraft {
+ return { turnId, step: 0, role: "assistant", type: "usage", data: u };
+}
+
+beforeAll(() => {
+ fakeDb = new FakeDatabase();
+});
+
+beforeEach(() => {
+ fakeDb.rows = [];
+});
+
+// ---------------------------------------------------------------------------
+// usage chunk persistence + side-channel invariants
+// ---------------------------------------------------------------------------
+describe("usage chunk rows (DB-backed)", () => {
+ const TAB = "tab-usage";
+
+ it("persists usage rows alongside content rows with contiguous seqs", () => {
+ appendChunks(TAB, [
+ { turnId: "t1", step: 0, role: "user", type: "text", data: { text: "hi" } },
+ { turnId: "t1", step: 0, role: "assistant", type: "text", data: { text: "yo" } },
+ usageDraft("t1", {
+ inputTokens: 100,
+ outputTokens: 10,
+ cacheReadTokens: 0,
+ cacheWriteTokens: 90,
+ }),
+ ]);
+ // All three rows landed with contiguous seqs.
+ expect(fakeDb.rows.map((r) => r.seq)).toEqual([0, 1, 2]);
+ expect(fakeDb.rows.map((r) => r.type)).toEqual(["text", "text", "usage"]);
+ });
+
+ it("excludes usage rows from getChunksForTab (all variants)", () => {
+ appendChunks(TAB, [
+ { turnId: "t1", step: 0, role: "user", type: "text", data: { text: "q" } },
+ usageDraft("t1", {
+ inputTokens: 100,
+ outputTokens: 10,
+ cacheReadTokens: 0,
+ cacheWriteTokens: 90,
+ }),
+ { turnId: "t1", step: 0, role: "assistant", type: "text", data: { text: "a" } },
+ usageDraft("t1", {
+ inputTokens: 200,
+ outputTokens: 20,
+ cacheReadTokens: 150,
+ cacheWriteTokens: 0,
+ }),
+ ]);
+
+ // no options
+ const all = getChunksForTab(TAB);
+ expect(all.every((r) => r.type !== "usage")).toBe(true);
+ expect(all.map((r) => r.type)).toEqual(["text", "text"]);
+
+ // limit only
+ const limited = getChunksForTab(TAB, { limit: 10 });
+ expect(limited.every((r) => r.type !== "usage")).toBe(true);
+ expect(limited).toHaveLength(2);
+
+ // before only — `before` is a seq cursor; usage seqs must never surface
+ const before = getChunksForTab(TAB, { before: 100 });
+ expect(before.every((r) => r.type !== "usage")).toBe(true);
+ expect(before).toHaveLength(2);
+
+ // before + limit
+ const bl = getChunksForTab(TAB, { before: 100, limit: 10 });
+ expect(bl.every((r) => r.type !== "usage")).toBe(true);
+ expect(bl).toHaveLength(2);
+ });
+
+ it("excludes usage rows from getTotalChunkCount", () => {
+ appendChunks(TAB, [
+ { turnId: "t1", step: 0, role: "user", type: "text", data: { text: "q" } },
+ { turnId: "t1", step: 0, role: "assistant", type: "text", data: { text: "a" } },
+ usageDraft("t1", {
+ inputTokens: 100,
+ outputTokens: 10,
+ cacheReadTokens: 0,
+ cacheWriteTokens: 90,
+ }),
+ ]);
+ // 3 rows total, but only 2 visible.
+ expect(getTotalChunkCount(TAB)).toBe(2);
+ });
+});
+
+// ---------------------------------------------------------------------------
+// getUsageStatsForTab — backend aggregate
+// ---------------------------------------------------------------------------
+describe("getUsageStatsForTab", () => {
+ const TAB = "tab-agg";
+
+ it("returns null when the tab has no usage rows", () => {
+ appendChunks(TAB, [
+ { turnId: "t1", step: 0, role: "assistant", type: "text", data: { text: "a" } },
+ ]);
+ expect(getUsageStatsForTab(TAB)).toBeNull();
+ });
+
+ it("sums cumulative tokens, counts requests, and reports the last request's split", () => {
+ appendChunks(TAB, [
+ usageDraft("t1", {
+ inputTokens: 1000,
+ outputTokens: 40,
+ cacheReadTokens: 0,
+ cacheWriteTokens: 900,
+ }),
+ usageDraft("t1", {
+ inputTokens: 1200,
+ outputTokens: 60,
+ cacheReadTokens: 1000,
+ cacheWriteTokens: 100,
+ }),
+ ]);
+
+ const stats = getUsageStatsForTab(TAB);
+ expect(stats).not.toBeNull();
+ expect(stats?.requests).toBe(2);
+ expect(stats?.inputTokens).toBe(2200);
+ expect(stats?.outputTokens).toBe(100);
+ expect(stats?.cacheReadTokens).toBe(1000);
+ expect(stats?.cacheWriteTokens).toBe(1000);
+ // `last` = the most recent (highest-seq) usage row.
+ expect(stats?.last).toEqual({
+ inputTokens: 1200,
+ outputTokens: 60,
+ cacheReadTokens: 1000,
+ cacheWriteTokens: 100,
+ });
+ });
+
+ it("is structurally identical to the frontend CacheStats shape (seeds directly)", () => {
+ appendChunks(TAB, [
+ usageDraft("t1", {
+ inputTokens: 5,
+ outputTokens: 1,
+ cacheReadTokens: 2,
+ cacheWriteTokens: 3,
+ }),
+ ]);
+ const stats = getUsageStatsForTab(TAB);
+ expect(Object.keys(stats ?? {}).sort()).toEqual(
+ [
+ "cacheReadTokens",
+ "cacheWriteTokens",
+ "inputTokens",
+ "last",
+ "outputTokens",
+ "requests",
+ ].sort(),
+ );
+ });
+
+ it("is scoped per tab", () => {
+ appendChunks("tab-a", [
+ usageDraft("t1", {
+ inputTokens: 10,
+ outputTokens: 1,
+ cacheReadTokens: 0,
+ cacheWriteTokens: 0,
+ }),
+ ]);
+ appendChunks("tab-b", [
+ usageDraft("t2", {
+ inputTokens: 20,
+ outputTokens: 2,
+ cacheReadTokens: 0,
+ cacheWriteTokens: 0,
+ }),
+ ]);
+ expect(getUsageStatsForTab("tab-a")?.inputTokens).toBe(10);
+ expect(getUsageStatsForTab("tab-b")?.inputTokens).toBe(20);
+ });
+});
diff --git a/packages/core/tests/models/catalog.test.ts b/packages/core/tests/models/catalog.test.ts
new file mode 100644
index 0000000..51043e6
--- /dev/null
+++ b/packages/core/tests/models/catalog.test.ts
@@ -0,0 +1,158 @@
+import { existsSync, rmSync, utimesSync, writeFileSync } from "node:fs";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+ __resetCatalogCacheForTests,
+ getModelsCatalog,
+ resolveContextLimit,
+} from "../../src/models/catalog.js";
+
+const CACHE_PATH = "/tmp/dispatch/models-dev.json";
+
+// A trimmed models.dev-shaped catalog covering the providers we support.
+const CATALOG = {
+ anthropic: {
+ id: "anthropic",
+ models: {
+ "claude-sonnet-4-5": { limit: { context: 200000, output: 64000 } },
+ "claude-sonnet-4-6": { limit: { context: 1000000, output: 64000 } },
+ },
+ },
+ opencode: {
+ id: "opencode",
+ models: {
+ "glm-4-6": { limit: { context: 131072, output: 8192 } },
+ },
+ },
+};
+
+function mockFetchOnce(catalog: unknown, ok = true, status = 200) {
+ const fn = vi.fn(() =>
+ Promise.resolve({
+ ok,
+ status,
+ text: () => Promise.resolve(JSON.stringify(catalog)),
+ } as Response),
+ );
+ vi.stubGlobal("fetch", fn);
+ return fn;
+}
+
+beforeEach(() => {
+ __resetCatalogCacheForTests();
+ if (existsSync(CACHE_PATH)) rmSync(CACHE_PATH);
+ delete process.env.DISPATCH_DISABLE_MODELS_FETCH;
+});
+
+afterEach(() => {
+ vi.unstubAllGlobals();
+ if (existsSync(CACHE_PATH)) rmSync(CACHE_PATH);
+});
+
+describe("resolveContextLimit", () => {
+ it("resolves a known anthropic model to its context window", async () => {
+ mockFetchOnce(CATALOG);
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBe(200000);
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-6")).toBe(1000000);
+ });
+
+ it("maps opencode-anthropic to the anthropic catalog, then opencode fallback", async () => {
+ mockFetchOnce(CATALOG);
+ // Present in the anthropic catalog.
+ expect(await resolveContextLimit("opencode-anthropic", "claude-sonnet-4-5")).toBe(200000);
+ // Absent in anthropic, found in the opencode gateway catalog.
+ expect(await resolveContextLimit("opencode-anthropic", "glm-4-6")).toBe(131072);
+ });
+
+ it("returns null for an unknown model id", async () => {
+ mockFetchOnce(CATALOG);
+ expect(await resolveContextLimit("anthropic", "no-such-model")).toBeNull();
+ });
+
+ it("returns null for an unsupported provider (no network needed)", async () => {
+ const fetchFn = mockFetchOnce(CATALOG);
+ expect(await resolveContextLimit("google", "gemini-2.5-pro")).toBeNull();
+ expect(await resolveContextLimit("anthropic", "")).toBeNull();
+ expect(fetchFn).not.toHaveBeenCalled();
+ });
+
+ it("returns null when the model has no positive context limit", async () => {
+ mockFetchOnce({
+ anthropic: { id: "anthropic", models: { broken: { limit: { context: 0 } } } },
+ });
+ expect(await resolveContextLimit("anthropic", "broken")).toBeNull();
+ });
+
+ it("does not throw on a malformed provider entry missing `models`", async () => {
+ // A provider object without a `models` map must degrade to null, not crash.
+ mockFetchOnce({ anthropic: { id: "anthropic" } });
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBeNull();
+ });
+
+ it("does not throw when limit/context fields are absent", async () => {
+ mockFetchOnce({ anthropic: { id: "anthropic", models: { m: {} } } });
+ expect(await resolveContextLimit("anthropic", "m")).toBeNull();
+ });
+});
+
+describe("getModelsCatalog caching", () => {
+ it("fetches once and serves the in-process memo on subsequent calls", async () => {
+ const fetchFn = mockFetchOnce(CATALOG);
+ await resolveContextLimit("anthropic", "claude-sonnet-4-5");
+ await resolveContextLimit("anthropic", "claude-sonnet-4-6");
+ await getModelsCatalog();
+ expect(fetchFn).toHaveBeenCalledTimes(1);
+ });
+
+ it("reuses a fresh disk cache without re-fetching across processes", async () => {
+ // Simulate another process having written a fresh cache.
+ writeFileSync(CACHE_PATH, JSON.stringify(CATALOG), "utf-8");
+ const fetchFn = vi.fn(() => Promise.reject(new Error("network should not be hit")));
+ vi.stubGlobal("fetch", fetchFn);
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBe(200000);
+ expect(fetchFn).not.toHaveBeenCalled();
+ });
+
+ it("falls back to a STALE disk cache when the network fails", async () => {
+ writeFileSync(CACHE_PATH, JSON.stringify(CATALOG), "utf-8");
+ // Age the cache well past the TTL so the fetch path is taken.
+ const old = Date.now() / 1000 - 3600;
+ utimesSync(CACHE_PATH, old, old);
+ const fetchFn = vi.fn(() => Promise.reject(new Error("offline")));
+ vi.stubGlobal("fetch", fetchFn);
+ const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBe(200000);
+ expect(fetchFn).toHaveBeenCalledTimes(1);
+ warn.mockRestore();
+ });
+
+ it("returns null when fetch fails and no cache exists", async () => {
+ const fetchFn = vi.fn(() => Promise.reject(new Error("offline")));
+ vi.stubGlobal("fetch", fetchFn);
+ const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBeNull();
+ warn.mockRestore();
+ });
+
+ it("does not hit the network when DISPATCH_DISABLE_MODELS_FETCH is set", async () => {
+ process.env.DISPATCH_DISABLE_MODELS_FETCH = "1";
+ const fetchFn = vi.fn(() => Promise.reject(new Error("should not fetch")));
+ vi.stubGlobal("fetch", fetchFn);
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBeNull();
+ expect(fetchFn).not.toHaveBeenCalled();
+ });
+
+ it("memoizes the fallback after a failed fetch so it does not re-hit the network", async () => {
+ const fetchFn = vi.fn(() => Promise.reject(new Error("offline")));
+ vi.stubGlobal("fetch", fetchFn);
+ const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+
+ // First lookup triggers the (failing) fetch.
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-5")).toBeNull();
+ // Subsequent lookups within the penalty window must NOT re-fetch.
+ expect(await resolveContextLimit("anthropic", "claude-sonnet-4-6")).toBeNull();
+ await getModelsCatalog();
+ expect(fetchFn).toHaveBeenCalledTimes(1);
+ warn.mockRestore();
+ });
+});