diff options
| author | Adam Malczewski <[email protected]> | 2026-05-30 20:06:31 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-05-30 20:06:31 +0900 |
| commit | 0f39b6f78957aacf206012ad2193d9b0c1940c1f (patch) | |
| tree | ff5f2da8b4f3cdf56cf50d44b8fec75a489ad6fe /packages/core/src/db/chunks.ts | |
| parent | 8c58a973b0d021689cebad5c0cc6d56956bbc2f6 (diff) | |
| download | dispatch-0f39b6f78957aacf206012ad2193d9b0c1940c1f.tar.gz dispatch-0f39b6f78957aacf206012ad2193d9b0c1940c1f.zip | |
refactor(chunks): append-only chunk log with per-step cache-stable wire
Replace the message-as-container model with a flat, append-only chunk log.
- chunks table (id, tab_id, seq, turn_id, step, role, type, data_json): one
row per chunk; tool_call (assistant) and tool_result (tool) are SEPARATE
rows linked by callId. Message/turn are derived groupings, not stored.
- chunks/transform.ts: DB-free explode (Chunk[] -> rows) / group (rows ->
messages), shared by backend and the browser frontend.
- Cache fix: toModelMessages segments each turn at tool-batch boundaries into
stable [assistant, tool] pairs per step, so earlier steps serialize
byte-identically across requests (kills the prompt-cache churn).
- agent-manager persists a turn's chunks on seal (once), discarding a failed
fallback attempt's partial chunks; rebuilds agent history from the log.
- GET /messages windows the log by chunk seq then groups; loadMoreMessages
merges a turn split across the window boundary by turnId.
- One-shot migration drops the legacy messages table and clears tabs;
settings/credentials/keys/usage preserved.
Full suite green (317 tests); biome, tsc, and svelte-check clean.
Diffstat (limited to 'packages/core/src/db/chunks.ts')
| -rw-r--r-- | packages/core/src/db/chunks.ts | 150 |
1 files changed, 150 insertions, 0 deletions
diff --git a/packages/core/src/db/chunks.ts b/packages/core/src/db/chunks.ts new file mode 100644 index 0000000..6841eb5 --- /dev/null +++ b/packages/core/src/db/chunks.ts @@ -0,0 +1,150 @@ +import { randomUUID } from "node:crypto"; +import { + explodeTurn, + explodeUserText, + groupRowsToMessages, + type MessageRow, +} from "../chunks/transform.js"; +import type { ChunkData, ChunkRow, ChunkRowDraft, TextData } from "../types/index.js"; +import { getDatabase } from "./index.js"; + +// Re-export the DB-free transforms so existing barrel consumers +// (`@dispatch/core`) keep importing them from here. The browser frontend deep- +// imports them directly from `chunks/transform.js` to avoid the DB dependency. +export { explodeTurn, explodeUserText, groupRowsToMessages, type MessageRow }; + +// ─── Persistence ───────────────────────────────────────────────── + +function mapRow(row: Record<string, unknown>): ChunkRow { + let data: ChunkData; + try { + data = JSON.parse(row.data_json as string) as ChunkData; + } catch { + data = { text: "" } as TextData; + } + return { + id: row.id as string, + tabId: row.tab_id as string, + seq: row.seq as number, + turnId: row.turn_id as string, + step: row.step as number, + role: row.role as ChunkRow["role"], + type: row.type as ChunkRow["type"], + data, + createdAt: row.created_at as number, + }; +} + +/** + * Append one or more chunk-row drafts to a tab, assigning a monotonic per-tab + * `seq` and a fresh id/timestamp to each. Returns the inserted rows in order. + */ +export function appendChunks(tabId: string, drafts: ChunkRowDraft[]): ChunkRow[] { + if (drafts.length === 0) return []; + const db = getDatabase(); + const maxSeq = db + .query("SELECT COALESCE(MAX(seq), -1) as max_seq FROM chunks WHERE tab_id = $tabId") + .get({ $tabId: tabId }) as { max_seq: number }; + let seq = (maxSeq?.max_seq ?? -1) + 1; + const now = Date.now(); + const insert = db.query( + `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)`, + ); + const out: ChunkRow[] = []; + for (const draft of drafts) { + const id = randomUUID(); + insert.run({ + $id: id, + $tabId: tabId, + $seq: seq, + $turnId: draft.turnId, + $step: draft.step, + $role: draft.role, + $type: draft.type, + $dataJson: JSON.stringify(draft.data), + $now: now, + }); + out.push({ + id, + tabId, + seq, + turnId: draft.turnId, + step: draft.step, + role: draft.role, + type: draft.type, + data: draft.data, + createdAt: now, + }); + seq++; + } + return out; +} + +/** + * Read chunk rows for a tab in `seq` order (ASC). Pagination mirrors the old + * message pagination but at chunk granularity: + * - no options → all rows; + * - `before` → rows with `seq < before`, most-recent-first then reversed; + * - `limit` → most recent `limit` rows, reversed to ASC. + */ +export function getChunksForTab( + tabId: string, + options?: { limit?: number; before?: number }, +): ChunkRow[] { + const db = getDatabase(); + if (!options) { + const rows = db + .query("SELECT * FROM chunks WHERE tab_id = $tabId ORDER BY seq ASC") + .all({ $tabId: tabId }) as Array<Record<string, unknown>>; + return rows.map(mapRow); + } + const { limit, before } = options; + if (before !== undefined) { + if (limit !== undefined) { + const rows = db + .query( + "SELECT * FROM chunks WHERE tab_id = $tabId 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") + .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") + .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") + .all({ $tabId: tabId }) as Array<Record<string, unknown>>; + return rows.map(mapRow); +} + +/** + * Derived, grouped view of a tab's full history as messages. Used to + * pre-populate the agent's in-memory `ChatMessage[]` history when an Agent is + * (re)constructed. Always reads the full log (grouping a partial window would + * be lossy for the rebuild path). + */ +export function getMessagesForTab(tabId: string): MessageRow[] { + return groupRowsToMessages(getChunksForTab(tabId)); +} + +export function getTotalChunkCount(tabId: string): number { + const db = getDatabase(); + const row = db + .query("SELECT COUNT(*) as count FROM chunks WHERE tab_id = $tabId") + .get({ $tabId: tabId }) as { count: number } | null; + return row?.count ?? 0; +} + +export function clearChunksForTab(tabId: string): void { + const db = getDatabase(); + db.query("DELETE FROM chunks WHERE tab_id = $tabId").run({ $tabId: tabId }); +} |
