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.

3 files changed, 188 insertions, 162 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 });
+}
diff --git a/packages/core/src/db/index.ts b/packages/core/src/db/index.ts
index e63b266..18dd1b5 100644
--- a/packages/core/src/db/index.ts
+++ b/packages/core/src/db/index.ts
@@ -93,16 +93,46 @@ export function getDatabase(): Database {
 		// Column already exists — ignore
 	}
 
-	_db.run(`CREATE TABLE IF NOT EXISTS messages (
-		id           TEXT PRIMARY KEY,
-		tab_id       TEXT NOT NULL REFERENCES tabs(id),
-		seq          INTEGER NOT NULL,
-		role         TEXT NOT NULL,
-		content_json TEXT NOT NULL,
-		created_at   INTEGER NOT NULL
+	// ─── Append-only chunk log (replaces the old `messages` blob table) ──
+	//
+	// A conversation is stored as a flat, append-only stream of chunk rows
+	// keyed by a per-tab monotonic `seq`. "Message" and "turn" are DERIVED
+	// groupings (see db/chunks.ts), never stored containers. This is what
+	// powers per-chunk frontend pagination AND the stable per-step wire
+	// format that fixes Anthropic prompt-cache churn (see plan-chunk-log.md).
+	//
+	//   role  : 'user' | 'assistant' | 'tool' | 'system'
+	//   type  : 'text' | 'thinking' | 'tool_call' | 'tool_result' | 'error' | 'system'
+	//   step  : LLM round-trip index within a turn (user/system rows = 0)
+	//   data_json: the type-specific payload (see ChunkData in types)
+	_db.run(`CREATE TABLE IF NOT EXISTS chunks (
+		id          TEXT PRIMARY KEY,
+		tab_id      TEXT NOT NULL,
+		seq         INTEGER NOT NULL,
+		turn_id     TEXT NOT NULL,
+		step        INTEGER NOT NULL DEFAULT 0,
+		role        TEXT NOT NULL,
+		type        TEXT NOT NULL,
+		data_json   TEXT NOT NULL,
+		created_at  INTEGER NOT NULL
 	)`);
 
-	_db.run(`CREATE INDEX IF NOT EXISTS idx_messages_tab ON messages(tab_id, seq)`);
+	_db.run(`CREATE INDEX IF NOT EXISTS idx_chunks_tab_seq ON chunks(tab_id, seq)`);
+
+	// One-shot migration off the legacy `messages` blob model. Beta software,
+	// no backward compatibility: the old chat history is destroyed (tabs +
+	// messages), while settings / credentials / api_keys / usage_cache /
+	// wake_schedule are preserved. Detect the old schema by the presence of
+	// the `messages` table; once dropped, this branch never runs again.
+	const hasLegacyMessages = _db
+		.query("SELECT name FROM sqlite_master WHERE type='table' AND name='messages'")
+		.get() as { name: string } | null;
+	if (hasLegacyMessages) {
+		_db.run("DROP TABLE IF EXISTS messages");
+		// Clear conversation containers too (fresh slate for the new model).
+		_db.run("DELETE FROM tabs");
+		_db.run("DELETE FROM chunks");
+	}
 
 	_db.run(`CREATE TABLE IF NOT EXISTS settings (
 		key   TEXT PRIMARY KEY,
diff --git a/packages/core/src/db/messages.ts b/packages/core/src/db/messages.ts
deleted file mode 100644
index 7fc6ccf..0000000
--- a/packages/core/src/db/messages.ts
+++ /dev/null
@@ -1,154 +0,0 @@
-import type { Chunk, MessageRole } from "../types/index.js";
-import { getDatabase } from "./index.js";
-
-/**
- * A persisted message row, with `content_json` already parsed into a `Chunk[]`.
- * Mirrors the new schema (no `thinking` column — that lived under the old
- * `content + toolCalls + toolResults + thinking` model).
- */
-export interface MessageRow {
-	id: string;
-	tabId: string;
-	seq: number;
-	role: MessageRole;
-	chunks: Chunk[];
-	createdAt: number;
-}
-
-/**
- * Append a new message to the tab. Caller passes the already-serialized
- * chunk list as `contentJson` (i.e. `JSON.stringify(chunks)`).
- */
-export function appendMessage(
-	tabId: string,
-	id: string,
-	role: MessageRole,
-	contentJson: string,
-): void {
-	const db = getDatabase();
-	const maxSeq = db
-		.query("SELECT COALESCE(MAX(seq), -1) as max_seq FROM messages WHERE tab_id = $tabId")
-		.get({ $tabId: tabId }) as { max_seq: number };
-	const seq = (maxSeq?.max_seq ?? -1) + 1;
-	db.query(
-		`INSERT INTO messages (id, tab_id, seq, role, content_json, created_at)
-		 VALUES ($id, $tabId, $seq, $role, $contentJson, $now)`,
-	).run({
-		$id: id,
-		$tabId: tabId,
-		$seq: seq,
-		$role: role,
-		$contentJson: contentJson,
-		$now: Date.now(),
-	});
-}
-
-/**
- * Replace the persisted chunks for an existing message. `contentJson` is
- * the already-serialized chunk list.
- */
-export function updateMessage(id: string, contentJson: string): void {
-	const db = getDatabase();
-	db.query("UPDATE messages SET content_json = $contentJson WHERE id = $id").run({
-		$id: id,
-		$contentJson: contentJson,
-	});
-}
-
-/**
- * Read messages for a tab in seq order (ASC). `content_json` is parsed into
- * `Chunk[]` here so callers don't have to. If a row's JSON is malformed,
- * the message is returned with an empty chunk list rather than throwing.
- *
- * When `options` is omitted, returns ALL messages (backward compatible).
- *
- * When `options.before` is provided, returns messages with `seq < before`,
- * taking the most recent ones first (DESC) up to `options.limit`, then
- * reversing back to ASC before returning.
- *
- * When only `options.limit` is provided, returns the most recent `limit`
- * messages, reversed back to ASC.
- */
-export function getMessagesForTab(
-	tabId: string,
-	options?: { limit?: number; before?: number },
-): MessageRow[] {
-	const db = getDatabase();
-
-	const mapRow = (row: Record<string, unknown>): MessageRow => {
-		const rawJson = row.content_json as string;
-		let chunks: Chunk[];
-		try {
-			const parsed = JSON.parse(rawJson);
-			chunks = Array.isArray(parsed) ? (parsed as Chunk[]) : [];
-		} catch {
-			chunks = [];
-		}
-		return {
-			id: row.id as string,
-			tabId: row.tab_id as string,
-			seq: row.seq as number,
-			role: row.role as MessageRole,
-			chunks,
-			createdAt: row.created_at as number,
-		};
-	};
-
-	// Backward-compatible path: no options → ALL messages, seq ASC.
-	if (!options) {
-		const rows = db
-			.query("SELECT * FROM messages WHERE tab_id = $tabId ORDER BY seq ASC")
-			.all({ $tabId: tabId }) as Array<Record<string, unknown>>;
-		return rows.map(mapRow);
-	}
-
-	const { limit, before } = options;
-
-	// Paginated path: fetch DESC, then reverse to ASC before returning.
-	if (before !== undefined) {
-		// `seq < before`, DESC, optionally limited.
-		if (limit !== undefined) {
-			const rows = db
-				.query(
-					"SELECT * FROM messages 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 messages 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();
-	}
-
-	// Only `limit` provided: most recent `limit`, reversed to ASC.
-	if (limit !== undefined) {
-		const rows = db
-			.query("SELECT * FROM messages 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();
-	}
-
-	// `options` was provided but empty → same as no options.
-	const rows = db
-		.query("SELECT * FROM messages WHERE tab_id = $tabId ORDER BY seq ASC")
-		.all({ $tabId: tabId }) as Array<Record<string, unknown>>;
-	return rows.map(mapRow);
-}
-
-/**
- * Return the total number of persisted messages for a tab.
- * Used by the API to advertise total history size alongside a paginated window.
- */
-export function getTotalMessageCount(tabId: string): number {
-	const db = getDatabase();
-	const row = db
-		.query("SELECT COUNT(*) as count FROM messages WHERE tab_id = $tabId")
-		.get({ $tabId: tabId }) as { count: number } | null;
-	return row?.count ?? 0;
-}
-
-export function clearMessagesForTab(tabId: string): void {
-	const db = getDatabase();
-	db.query("DELETE FROM messages WHERE tab_id = $tabId").run({ $tabId: tabId });
-}