From 3a999e074ba62a90d4a335d06eed6baf60787331 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 03:56:50 -0800 Subject: [PATCH 1/5] perf(server): Load earlier turns no longer reads every payload of a long thread A history page indexed the whole thread by fetching each turn item row and node, and ran four LIKE patterns over every tool payload. On a 45,000 item thread that is 506 MB of disk reads from a cold cache, so the client's 6 s limit ended the request first. Migration 084 adds two covering indexes for the index read. The inline output scan now covers only the runs the page spans. Co-Authored-By: Claude Fable 5.2 --- .../orchestration-v2/ProjectionStore.test.ts | 24 +++- .../src/orchestration-v2/ProjectionStore.ts | 110 +++++++++++++----- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/086_TimelineIndexReads.ts | 19 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 5 files changed, 127 insertions(+), 30 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index bcd6732c23782..b0b387718d1ff 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3770,6 +3770,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { window.projection.visibleTurnItems.find((row) => row.item.deferredWork)?.item.ordinal, 3, ); + const pageRead = traceSqlStatements(); const page = yield* historyPage( threadId, encodeThreadHistoryCursor({ @@ -3781,8 +3782,29 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { undefined, false, "deferred", - ); + ).pipe(Effect.withTracer(pageRead.tracer)); assert.isUndefined(page.items.find((row) => row.item.ordinal === 2)?.item.deferredWork); + // Tool payloads hold megabytes, so a page looks for inline output only in its own runs + // and indexes the thread without fetching a row or node per item. + const payloadScans = pageRead.statements.filter((statement) => + statement.includes("LIKE '%htmlRender%'"), + ); + assert.lengthOf(payloadScans, 1); + assert.include(payloadScans[0], "item.run_id = json_extract(wanted.value, '$.id')"); + const indexRead = pageRead.statements.find((statement) => + statement.includes("ORDER BY item.ordinal ASC, item.turn_item_id ASC"), + ); + assert.isDefined(indexRead); + const indexPlan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${indexRead}`, + ); + assert.includeMembers( + indexPlan.map((row) => row.detail), + [ + "SEARCH item USING INDEX orchestration_v2_projection_turn_items_timeline_idx (thread_id=?)", + "SEARCH node USING COVERING INDEX orchestration_v2_projection_nodes_root_idx (node_id=?) LEFT-JOIN", + ], + ); } yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'command_execution', payload_json = json_set(payload_json, '$.type', 'command_execution') WHERE thread_id = ${threadId} AND ordinal = 2`; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2bf1e896b4973..0fe9fe75b4206 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -103,6 +103,7 @@ import { THREAD_HISTORY_MAX_RAW_TURNS, missingInterruptRequests, DEFERRED_WORK_ITEM_TYPES, + decodeThreadHistoryCursor, deferWorkHistoryRows, selectDeferredWorkPage, encodeThreadHistoryCursor, @@ -6194,12 +6195,14 @@ export const layer: Layer.Layer = const readTimelineIndex: ( threadId: ThreadId, seen: ReadonlySet, - includePersistentOutput?: boolean, + includeAttempts?: boolean, ) => Effect.Effect = Effect.fn( "ProjectionStore.readTimelineIndex", )( - function* (threadId: ThreadId, seen: ReadonlySet, includePersistentOutput = false) { + function* (threadId: ThreadId, seen: ReadonlySet, includeAttempts = false) { const records = yield* getThreadRecords(threadId, ["runs", "attempts"]); + // SQLite does not pick the covering indexes of migration 086 by itself, and without them + // this read fetches every row of the thread. It reads a payload only for a message. const rows = yield* sql<{ turn_item_id: string; run_id: string | null; @@ -6211,22 +6214,20 @@ export const layer: Layer.Layer = message_id: string | null; status: OrchestrationV2TurnItem["status"]; ordinal: number; - persistent_output: number; attempt_id: string | null; attempt_status: string | null; }>`SELECT item.turn_item_id, item.run_id, item.node_id, item.parent_item_id, item.type, item.status, item.ordinal, - ${includePersistentOutput ? sql`attempt.attempt_id` : sql`NULL`} AS attempt_id, - ${includePersistentOutput ? sql`attempt.status` : sql`NULL`} AS attempt_status, - ${ - includePersistentOutput ? sql`CASE WHEN ${inlineWorkOutput} THEN 1 ELSE 0 END` : sql`0` - } AS persistent_output, + ${includeAttempts ? sql`attempt.attempt_id` : sql`NULL`} AS attempt_id, + ${includeAttempts ? sql`attempt.status` : sql`NULL`} AS attempt_status, CASE WHEN item.type = 'user_message' THEN json_extract(item.payload_json, '$.inputIntent') END AS input_intent, CASE WHEN item.type = 'user_message' THEN json_extract(item.payload_json, '$.createdBy') END AS created_by, CASE WHEN item.type IN ('user_message', 'assistant_message') THEN json_extract(item.payload_json, '$.messageId') END AS message_id FROM orchestration_v2_projection_turn_items AS item + INDEXED BY orchestration_v2_projection_turn_items_timeline_idx ${ - includePersistentOutput - ? sql`LEFT JOIN orchestration_v2_projection_nodes AS node ON node.node_id = item.node_id + includeAttempts + ? sql`LEFT JOIN orchestration_v2_projection_nodes AS node + INDEXED BY orchestration_v2_projection_nodes_root_idx ON node.node_id = item.node_id LEFT JOIN orchestration_v2_projection_run_attempts AS attempt ON attempt.attempt_id = ( SELECT attempt_id FROM orchestration_v2_projection_run_attempts WHERE run_id = item.run_id AND root_node_id = COALESCE(node.root_node_id, item.node_id) @@ -6249,7 +6250,6 @@ export const layer: Layer.Layer = runStatus: row.run_id === null ? undefined : runStatuses.get(row.run_id), attemptId: row.attempt_id, attemptStatus: row.attempt_status, - persistentOutput: row.persistent_output === 1, ...(row.input_intent === null ? {} : { inputIntent: row.input_intent }), ...(row.created_by === null ? {} : { createdBy: row.created_by }), ...(row.message_id === null ? {} : { messageId: MessageId.make(row.message_id) }), @@ -6272,7 +6272,7 @@ export const layer: Layer.Layer = const source = yield* readTimelineIndex( fork.threadId, new Set([...seen, threadId]), - includePersistentOutput, + includeAttempts, ); const sourceRun = source.records.runs.find((run) => run.id === fork.runId); let inherited: Array = []; @@ -6343,14 +6343,14 @@ export const layer: Layer.Layer = Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })), ); const readSnapshotTimelineIndex = Effect.fn("ProjectionStore.readSnapshotTimelineIndex")( - function* (threadId: ThreadId, includePersistentOutput: boolean) { - const key = `${includePersistentOutput ? "work" : "plain"}:${threadId}`; + function* (threadId: ThreadId, includeAttempts: boolean) { + const key = `${includeAttempts ? "work" : "plain"}:${threadId}`; const cached = timelineIndexes.get(key); const index = cached !== undefined && (yield* latestThreadEventSequence(threadId, cached.index.threadIds)) === cached.sequence ? cached.index - : yield* readTimelineIndex(threadId, new Set(), includePersistentOutput); + : yield* readTimelineIndex(threadId, new Set(), includeAttempts); const sequence = index === cached?.index ? cached.sequence @@ -6392,32 +6392,84 @@ export const layer: Layer.Layer = position, canonicalPosition: position, })); - const historyItems = workMode === "deferred" ? deferWorkHistoryRows(indexed) : indexed; - const selection = yield* Effect.try({ + const { olderPolicy, cursorPosition } = yield* Effect.try({ try: () => { - const policy = olderHistoryPagePolicy({ items: indexed, cursor, throughEntryId }); - const page = + const { st, si } = decodeThreadHistoryCursor(cursor); + return { + olderPolicy: olderHistoryPagePolicy({ items: indexed, cursor, throughEntryId }), + cursorPosition: indexed.findLastIndex( + (row) => row.sourceThreadId === st && row.sourceItemId === si, + ), + }; + }, + catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), + }); + const selectPage = (rows: typeof indexed) => + Effect.try({ + try: () => workMode === "work" ? selectDeferredWorkPage({ - items: indexed, + items: rows, cursor, throughEntryId, snapshotSequence, rowEncodedBytes: () => 0, }) : selectHistoryPageFromCursor({ - items: historyItems, + items: workMode === "deferred" ? deferWorkHistoryRows(rows) : rows, cursor, snapshotSequence, rowEncodedBytes: () => 0, - policy, - }); - return { policy, page }; - }, - catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), - }); - const { page } = selection; - const policy = workMode === "work" ? THREAD_HISTORY_PAGE_POLICY : selection.policy; + policy: olderPolicy, + }), + catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), + }); + let page = yield* selectPage(indexed); + // Inline output is found by scanning payloads, which costs a whole long thread several + // seconds. Scan only the runs the page spans, then select again with what they hold. + const scannedRuns = new Set(); + const persistentOutputs = new Set(); + while (workMode !== undefined && page.items.length > 0) { + const runs = new Map( + indexed + .slice( + page.items[0]!.canonicalPosition, + Math.max(page.items.at(-1)!.canonicalPosition, cursorPosition) + 1, + ) + .flatMap(({ sourceThreadId, item }) => + item.runId !== null && + (item.runStatus === "completed" || item.attemptStatus === "superseded") && + !scannedRuns.has(`${sourceThreadId}:${item.runId}`) + ? [ + [ + `${sourceThreadId}:${item.runId}`, + { threadId: sourceThreadId, id: item.runId }, + ] as const, + ] + : [], + ), + ); + if (runs.size === 0) break; + for (const run of runs.keys()) scannedRuns.add(run); + const found = yield* sql<{ thread_id: string; turn_item_id: string }>` + SELECT item.thread_id, item.turn_item_id + FROM json_each(${yield* encodeTimelineSources([...runs.values()])}) AS wanted + CROSS JOIN orchestration_v2_projection_turn_items AS item + ON item.thread_id = json_extract(wanted.value, '$.threadId') + AND item.run_id = json_extract(wanted.value, '$.id') + WHERE item.status = 'completed' AND ${inlineWorkOutput}`; + if (found.length === 0) break; + for (const row of found) + persistentOutputs.add(`${row.thread_id}:${row.turn_item_id}`); + page = yield* selectPage( + indexed.map((row) => + persistentOutputs.has(`${row.sourceThreadId}:${row.sourceItemId}`) + ? { ...row, item: { ...row.item, persistentOutput: true } } + : row, + ), + ); + } + const policy = workMode === "work" ? THREAD_HISTORY_PAGE_POLICY : olderPolicy; const end = (page.items.at(-1)?.canonicalPosition ?? -1) + 1; const hasUserTurns = index.visible .slice(0, end) diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 37b74d5c6db01..6f5b7846cbaa3 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -88,6 +88,7 @@ import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; +import Migration0086 from "./Migrations/086_TimelineIndexReads.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -176,6 +177,7 @@ const migrationEntries = [ [83, "CommandReadIndexes", Migration0083], [84, "LastExecutedRunIndex", Migration0084], [85, "TurnItemCounts", Migration0085], + [86, "TimelineIndexReads", Migration0086], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts b/apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts new file mode 100644 index 0000000000000..9e235fae3411c --- /dev/null +++ b/apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts @@ -0,0 +1,19 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // A history page indexes every item of its thread. These carry the columns that read needs, so + // it stays in the indexes instead of fetching each row and node next to megabytes of payload. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_timeline_idx + ON orchestration_v2_projection_turn_items( + thread_id, ordinal, turn_item_id, run_id, node_id, parent_item_id, type, status + ) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_nodes_root_idx + ON orchestration_v2_projection_nodes(node_id, root_node_id) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 3c2cb075cfab4..16d4165a8b4b2 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -65,6 +65,7 @@ describe("V2 preview upgrade", () => { [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], [85, "TurnItemCounts"], + [86, "TimelineIndexReads"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -173,6 +174,7 @@ describe("V2 preview upgrade", () => { [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], [85, "TurnItemCounts"], + [86, "TimelineIndexReads"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 86d9e5c22c4d46cf1d9c263e2658216161c1ce4f Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 04:02:04 -0800 Subject: [PATCH 2/5] fix(server): a history page scans the cursor row's run and every work-mode run Co-Authored-By: Claude Fable 5.2 --- .../orchestration-v2/ProjectionStore.test.ts | 22 ++++++++++++++++++- .../src/orchestration-v2/ProjectionStore.ts | 13 +++++++---- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index b0b387718d1ff..aaba3e82868fe 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3806,8 +3806,28 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ], ); } + // Until its run is scanned, a cursor on inline output is hidden inside a work group. + yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'dynamic_tool', + payload_json = json_set(payload_json, '$.type', 'dynamic_tool', '$.toolName', 't3-code.delegate_task') + WHERE thread_id = ${threadId} AND ordinal = 4`; + const beforeInlineOutput = yield* historyPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: "turn-item:bounded-sql-history:4", + position: 0, + }), + undefined, + false, + "deferred", + ); + assert.deepEqual( + beforeInlineOutput.items.map((row) => row.item.ordinal), + [1, 2, 3], + ); yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'command_execution', - payload_json = json_set(payload_json, '$.type', 'command_execution') WHERE thread_id = ${threadId} AND ordinal = 2`; + payload_json = json_set(payload_json, '$.type', 'command_execution') WHERE thread_id = ${threadId} AND ordinal IN (2, 4)`; const sharedRunId = "run:bounded-sql-history:1"; for (const [ordinal, status] of [ [0, "completed"], diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 0fe9fe75b4206..7e081fc5d36f5 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6429,16 +6429,21 @@ export const layer: Layer.Layer = // seconds. Scan only the runs the page spans, then select again with what they hold. const scannedRuns = new Set(); const persistentOutputs = new Set(); - while (workMode !== undefined && page.items.length > 0) { + while (workMode !== undefined) { + // A cursor row hidden in a work group leaves the page empty, so its run counts too. + const from = page.items[0]?.canonicalPosition ?? cursorPosition; + if (from < 0) break; const runs = new Map( indexed .slice( - page.items[0]!.canonicalPosition, - Math.max(page.items.at(-1)!.canonicalPosition, cursorPosition) + 1, + from, + Math.max(page.items.at(-1)?.canonicalPosition ?? from, cursorPosition) + 1, ) .flatMap(({ sourceThreadId, item }) => item.runId !== null && - (item.runStatus === "completed" || item.attemptStatus === "superseded") && + (workMode === "work" || + item.runStatus === "completed" || + item.attemptStatus === "superseded") && !scannedRuns.has(`${sourceThreadId}:${item.runId}`) ? [ [ From 91a5afdb6265eb02a718577a250ab521d6da11f3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 04:06:30 -0800 Subject: [PATCH 3/5] fix(server): a history page scans its cursor row's run when a stale position moves the page Co-Authored-By: Claude Fable 5.2 --- .../orchestration-v2/ProjectionStore.test.ts | 49 ++++++++++++------- .../src/orchestration-v2/ProjectionStore.ts | 14 +++--- 2 files changed, 39 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index aaba3e82868fe..e39c67274a58a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3810,25 +3810,40 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'dynamic_tool', payload_json = json_set(payload_json, '$.type', 'dynamic_tool', '$.toolName', 't3-code.delegate_task') WHERE thread_id = ${threadId} AND ordinal = 4`; - const beforeInlineOutput = yield* historyPage( - threadId, - encodeThreadHistoryCursor({ - snapshotSequence: 0, - sourceThreadId: threadId, - sourceItemId: "turn-item:bounded-sql-history:4", - position: 0, - }), - undefined, - false, - "deferred", - ); - assert.deepEqual( - beforeInlineOutput.items.map((row) => row.item.ordinal), - [1, 2, 3], - ); + const sharedRunId = "run:bounded-sql-history:1"; + // A stale position selects from the cursor row itself, or from more than a page after it + // once the later rows are another run's and cannot fold. + for (const [position, laterStatus, laterRun] of [ + [0, "completed", sharedRunId], + [999, "running", "run:bounded-sql-history:300"], + ] as const) { + yield* sql`UPDATE orchestration_v2_projection_turn_items SET status = ${laterStatus}, + run_id = ${laterRun}, + payload_json = json_set(payload_json, '$.status', ${laterStatus}, '$.runId', ${laterRun}) + WHERE thread_id = ${threadId} AND ordinal >= 300`; + const beforeInlineOutput = yield* historyPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: "turn-item:bounded-sql-history:4", + position, + }), + undefined, + false, + "deferred", + ); + assert.deepEqual( + beforeInlineOutput.items.map((row) => row.item.ordinal), + [1, 2, 3], + ); + } + yield* sql`UPDATE orchestration_v2_projection_turn_items SET status = 'completed', + run_id = ${sharedRunId}, + payload_json = json_set(payload_json, '$.status', 'completed', '$.runId', ${sharedRunId}) + WHERE thread_id = ${threadId} AND ordinal >= 300`; yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'command_execution', payload_json = json_set(payload_json, '$.type', 'command_execution') WHERE thread_id = ${threadId} AND ordinal IN (2, 4)`; - const sharedRunId = "run:bounded-sql-history:1"; for (const [ordinal, status] of [ [0, "completed"], [1, "superseded"], diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7e081fc5d36f5..1a824879d1efd 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6430,15 +6430,15 @@ export const layer: Layer.Layer = const scannedRuns = new Set(); const persistentOutputs = new Set(); while (workMode !== undefined) { - // A cursor row hidden in a work group leaves the page empty, so its run counts too. - const from = page.items[0]?.canonicalPosition ?? cursorPosition; - if (from < 0) break; + // A cursor row hidden in a work group moves the page, so the scan always covers it. + const span = [...page.items.slice(0, 1), ...page.items.slice(-1)].map( + (row) => row.canonicalPosition, + ); + if (cursorPosition >= 0) span.push(cursorPosition); + if (span.length === 0) break; const runs = new Map( indexed - .slice( - from, - Math.max(page.items.at(-1)?.canonicalPosition ?? from, cursorPosition) + 1, - ) + .slice(Math.min(...span), Math.max(...span) + 1) .flatMap(({ sourceThreadId, item }) => item.runId !== null && (workMode === "work" || From af3c1826016cb40763f3a72e8574af8ebc667adb Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 04:17:33 -0800 Subject: [PATCH 4/5] fix(server): a history page scans up to the position a missing cursor row falls back to Co-Authored-By: Claude Fable 5.2 --- .../orchestration-v2/ProjectionStore.test.ts | 29 +++++++++++++++++-- .../src/orchestration-v2/ProjectionStore.ts | 20 ++++++------- .../orchestration-v2/threadHistoryPaging.ts | 2 +- 3 files changed, 38 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index e39c67274a58a..1d58da1cf97e8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3838,12 +3838,37 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { [1, 2, 3], ); } + // A cursor whose row is gone falls back to its position. Inline output between the page + // and that position can belong to a run with no other row there. + yield* sql`UPDATE orchestration_v2_projection_turn_items + SET status = CASE WHEN ordinal <= 700 THEN 'running' ELSE 'completed' END, + run_id = CASE WHEN ordinal <= 700 THEN 'run:bounded-sql-history:300' ELSE ${sharedRunId} END, + payload_json = json_set(payload_json, + '$.status', CASE WHEN ordinal <= 700 THEN 'running' ELSE 'completed' END, + '$.runId', CASE WHEN ordinal <= 700 THEN 'run:bounded-sql-history:300' ELSE ${sharedRunId} END) + WHERE thread_id = ${threadId} AND ordinal >= 100`; + yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'dynamic_tool', + payload_json = json_set(payload_json, '$.type', 'dynamic_tool', '$.toolName', 't3-code.delegate_task') + WHERE thread_id = ${threadId} AND ordinal = 720`; + const beforeMissingCursor = yield* historyPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: "turn-item:bounded-sql-history:gone", + position: 800, + }), + undefined, + false, + "deferred", + ); + assert.strictEqual(beforeMissingCursor.items.at(-1)?.item.ordinal, 720); yield* sql`UPDATE orchestration_v2_projection_turn_items SET status = 'completed', run_id = ${sharedRunId}, payload_json = json_set(payload_json, '$.status', 'completed', '$.runId', ${sharedRunId}) - WHERE thread_id = ${threadId} AND ordinal >= 300`; + WHERE thread_id = ${threadId} AND ordinal >= 100`; yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'command_execution', - payload_json = json_set(payload_json, '$.type', 'command_execution') WHERE thread_id = ${threadId} AND ordinal IN (2, 4)`; + payload_json = json_set(payload_json, '$.type', 'command_execution') WHERE thread_id = ${threadId} AND ordinal IN (2, 4, 720)`; for (const [ordinal, status] of [ [0, "completed"], [1, "superseded"], diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 1a824879d1efd..b0aaa9ca207a4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -105,6 +105,7 @@ import { DEFERRED_WORK_ITEM_TYPES, decodeThreadHistoryCursor, deferWorkHistoryRows, + findCursorEndIndex, selectDeferredWorkPage, encodeThreadHistoryCursor, olderHistoryPagePolicy, @@ -6393,15 +6394,14 @@ export const layer: Layer.Layer = canonicalPosition: position, })); const { olderPolicy, cursorPosition } = yield* Effect.try({ - try: () => { - const { st, si } = decodeThreadHistoryCursor(cursor); - return { - olderPolicy: olderHistoryPagePolicy({ items: indexed, cursor, throughEntryId }), - cursorPosition: indexed.findLastIndex( - (row) => row.sourceThreadId === st && row.sourceItemId === si, - ), - }; - }, + try: () => ({ + olderPolicy: olderHistoryPagePolicy({ items: indexed, cursor, throughEntryId }), + // The cursor's own row, or the row its stale position falls back to. + cursorPosition: Math.min( + findCursorEndIndex(indexed, decodeThreadHistoryCursor(cursor)), + indexed.length - 1, + ), + }), catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), }); const selectPage = (rows: typeof indexed) => @@ -6430,7 +6430,7 @@ export const layer: Layer.Layer = const scannedRuns = new Set(); const persistentOutputs = new Set(); while (workMode !== undefined) { - // A cursor row hidden in a work group moves the page, so the scan always covers it. + // Rows hidden in a work group move the page, so the scan always reaches the cursor. const span = [...page.items.slice(0, 1), ...page.items.slice(-1)].map( (row) => row.canonicalPosition, ); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 37b653c451fce..b79038a77a803 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -394,7 +394,7 @@ export function selectRecentTimelineWindow(input: { }); } -function findCursorEndIndex( +export function findCursorEndIndex( items: ReadonlyArray, cursor: ThreadHistoryCursorPayload, ): number { From 584dfd1dfa0557c07a51796244ece40808aff23d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 06:08:33 -0800 Subject: [PATCH 5/5] fix(server): history pages scan inline output per run without migration 086 The covering indexes of migration 086 are removed. They grew every database by about 6% and their build blocks the start of the server, while the scan of only the runs a page spans gives most of the gain alone. A work page now scans one run, the run of its rows. That includes rows that have no run, which the scan skipped before. Co-Authored-By: Claude Fable 5.2 --- .../orchestration-v2/ProjectionStore.test.ts | 41 ++++++++++-------- .../src/orchestration-v2/ProjectionStore.ts | 43 +++++++++---------- apps/server/src/persistence/Migrations.ts | 2 - .../Migrations/086_TimelineIndexReads.ts | 19 -------- .../reconcileV2PreviewMigration.test.ts | 2 - 5 files changed, 44 insertions(+), 63 deletions(-) delete mode 100644 apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 8b68ae6f1cedb..0df805ad73ac5 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -4219,33 +4219,40 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { "deferred", ).pipe(Effect.withTracer(pageRead.tracer)); assert.isUndefined(page.items.find((row) => row.item.ordinal === 2)?.item.deferredWork); - // Tool payloads hold megabytes, so a page looks for inline output only in its own runs - // and indexes the thread without fetching a row or node per item. + // Tool payloads hold megabytes, so a page looks for inline output only in its own runs. const payloadScans = pageRead.statements.filter((statement) => statement.includes("LIKE '%htmlRender%'"), ); assert.lengthOf(payloadScans, 1); - assert.include(payloadScans[0], "item.run_id = json_extract(wanted.value, '$.id')"); - const indexRead = pageRead.statements.find((statement) => - statement.includes("ORDER BY item.ordinal ASC, item.turn_item_id ASC"), - ); - assert.isDefined(indexRead); - const indexPlan = yield* sql.unsafe<{ readonly detail: string }>( - `EXPLAIN QUERY PLAN ${indexRead}`, - ); - assert.includeMembers( - indexPlan.map((row) => row.detail), - [ - "SEARCH item USING INDEX orchestration_v2_projection_turn_items_timeline_idx (thread_id=?)", - "SEARCH node USING COVERING INDEX orchestration_v2_projection_nodes_root_idx (node_id=?) LEFT-JOIN", - ], - ); + assert.include(payloadScans[0], "item.run_id IS json_extract(wanted.value, '$.id')"); } // Until its run is scanned, a cursor on inline output is hidden inside a work group. yield* sql`UPDATE orchestration_v2_projection_turn_items SET type = 'dynamic_tool', payload_json = json_set(payload_json, '$.type', 'dynamic_tool', '$.toolName', 't3-code.delegate_task') WHERE thread_id = ${threadId} AND ordinal = 4`; const sharedRunId = "run:bounded-sql-history:1"; + // A work page of rows that have no run leaves their inline output out too. + yield* sql`UPDATE orchestration_v2_projection_turn_items SET run_id = NULL + WHERE thread_id = ${threadId} AND ordinal BETWEEN 3 AND 6`; + const runlessWork = yield* historyPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: "turn-item:bounded-sql-history:6", + position: 6, + inclusive: true, + }), + "turn-item:bounded-sql-history:3", + false, + "work", + ); + assert.deepEqual( + runlessWork.items.map((row) => row.item.ordinal), + [3, 5, 6], + ); + yield* sql`UPDATE orchestration_v2_projection_turn_items SET run_id = ${sharedRunId} + WHERE thread_id = ${threadId} AND ordinal BETWEEN 3 AND 6`; // A stale position selects from the cursor row itself, or from more than a page after it // once the later rows are another run's and cannot fold. for (const [position, laterStatus, laterRun] of [ diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index bbea90bbb2f1f..2774d8936e777 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1133,7 +1133,7 @@ type ShellItemCountRow = { const encodeTimelineSources = Schema.encodeEffect( Schema.fromJsonString( - Schema.Array(Schema.Struct({ threadId: Schema.String, id: Schema.String })), + Schema.Array(Schema.Struct({ threadId: Schema.String, id: Schema.NullOr(Schema.String) })), ), ); const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema.String))); @@ -6281,8 +6281,6 @@ export const layer: Layer.Layer = )( function* (threadId: ThreadId, seen: ReadonlySet, includeAttempts = false) { const records = yield* getThreadRecords(threadId, ["runs", "attempts"]); - // SQLite does not pick the covering indexes of migration 086 by itself, and without them - // this read fetches every row of the thread. It reads a payload only for a message. const rows = yield* sql<{ turn_item_id: string; run_id: string | null; @@ -6303,11 +6301,9 @@ export const layer: Layer.Layer = CASE WHEN item.type = 'user_message' THEN json_extract(item.payload_json, '$.createdBy') END AS created_by, CASE WHEN item.type IN ('user_message', 'assistant_message') THEN json_extract(item.payload_json, '$.messageId') END AS message_id FROM orchestration_v2_projection_turn_items AS item - INDEXED BY orchestration_v2_projection_turn_items_timeline_idx ${ includeAttempts - ? sql`LEFT JOIN orchestration_v2_projection_nodes AS node - INDEXED BY orchestration_v2_projection_nodes_root_idx ON node.node_id = item.node_id + ? sql`LEFT JOIN orchestration_v2_projection_nodes AS node ON node.node_id = item.node_id LEFT JOIN orchestration_v2_projection_run_attempts AS attempt ON attempt.attempt_id = ( SELECT attempt_id FROM orchestration_v2_projection_run_attempts WHERE run_id = item.run_id AND root_node_id = COALESCE(node.root_node_id, item.node_id) @@ -6516,22 +6512,23 @@ export const layer: Layer.Layer = if (cursorPosition >= 0) span.push(cursorPosition); if (span.length === 0) break; const runs = new Map( - indexed - .slice(Math.min(...span), Math.max(...span) + 1) - .flatMap(({ sourceThreadId, item }) => - item.runId !== null && - (workMode === "work" || - item.runStatus === "completed" || - item.attemptStatus === "superseded") && - !scannedRuns.has(`${sourceThreadId}:${item.runId}`) - ? [ - [ - `${sourceThreadId}:${item.runId}`, - { threadId: sourceThreadId, id: item.runId }, - ] as const, - ] - : [], - ), + // A work page holds rows of one run, which can be the rows that have no run. + (workMode === "work" + ? page.items.slice(0, 1) + : indexed.slice(Math.min(...span), Math.max(...span) + 1) + ).flatMap(({ sourceThreadId, item }) => + (workMode === "work" || + (item.runId !== null && + (item.runStatus === "completed" || item.attemptStatus === "superseded"))) && + !scannedRuns.has(`${sourceThreadId}:${item.runId}`) + ? [ + [ + `${sourceThreadId}:${item.runId}`, + { threadId: sourceThreadId, id: item.runId }, + ] as const, + ] + : [], + ), ); if (runs.size === 0) break; for (const run of runs.keys()) scannedRuns.add(run); @@ -6540,7 +6537,7 @@ export const layer: Layer.Layer = FROM json_each(${yield* encodeTimelineSources([...runs.values()])}) AS wanted CROSS JOIN orchestration_v2_projection_turn_items AS item ON item.thread_id = json_extract(wanted.value, '$.threadId') - AND item.run_id = json_extract(wanted.value, '$.id') + AND item.run_id IS json_extract(wanted.value, '$.id') WHERE item.status = 'completed' AND ${inlineWorkOutput}`; if (found.length === 0) break; for (const row of found) diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6f5b7846cbaa3..37b74d5c6db01 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -88,7 +88,6 @@ import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; -import Migration0086 from "./Migrations/086_TimelineIndexReads.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -177,7 +176,6 @@ const migrationEntries = [ [83, "CommandReadIndexes", Migration0083], [84, "LastExecutedRunIndex", Migration0084], [85, "TurnItemCounts", Migration0085], - [86, "TimelineIndexReads", Migration0086], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts b/apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts deleted file mode 100644 index 9e235fae3411c..0000000000000 --- a/apps/server/src/persistence/Migrations/086_TimelineIndexReads.ts +++ /dev/null @@ -1,19 +0,0 @@ -import * as Effect from "effect/Effect"; -import * as SqlClient from "effect/sql/SqlClient"; - -export default Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - // A history page indexes every item of its thread. These carry the columns that read needs, so - // it stays in the indexes instead of fetching each row and node next to megabytes of payload. - yield* sql` - CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_timeline_idx - ON orchestration_v2_projection_turn_items( - thread_id, ordinal, turn_item_id, run_id, node_id, parent_item_id, type, status - ) - `; - yield* sql` - CREATE INDEX IF NOT EXISTS orchestration_v2_projection_nodes_root_idx - ON orchestration_v2_projection_nodes(node_id, root_node_id) - `; -}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 16d4165a8b4b2..3c2cb075cfab4 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -65,7 +65,6 @@ describe("V2 preview upgrade", () => { [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], [85, "TurnItemCounts"], - [86, "TimelineIndexReads"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -174,7 +173,6 @@ describe("V2 preview upgrade", () => { [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], [85, "TurnItemCounts"], - [86, "TimelineIndexReads"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );