diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index ebafef4bc5621..3f48cbf247fc7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -4205,6 +4205,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({ @@ -4216,12 +4217,100 @@ 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. + const payloadScans = pageRead.statements.filter((statement) => + statement.includes("LIKE '%htmlRender%'"), + ); + assert.lengthOf(payloadScans, 1); + assert.include(payloadScans[0], "item.run_id IS json_extract(wanted.value, '$.id')"); } - 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`; + // 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 [ + [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], + ); + } + // 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 >= 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, 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 101521c7e1064..4ccb6bbf85ad8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -103,7 +103,9 @@ import { THREAD_HISTORY_MAX_RAW_TURNS, missingInterruptRequests, DEFERRED_WORK_ITEM_TYPES, + decodeThreadHistoryCursor, deferWorkHistoryRows, + findCursorEndIndex, selectDeferredWorkPage, encodeThreadHistoryCursor, olderHistoryPagePolicy, @@ -1131,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))); @@ -6280,11 +6282,11 @@ 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"]); const rows = yield* sql<{ turn_item_id: string; @@ -6297,21 +6299,17 @@ 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 ${ - includePersistentOutput + includeAttempts ? 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 @@ -6335,7 +6333,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) }), @@ -6358,7 +6355,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 = []; @@ -6429,14 +6426,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 @@ -6478,32 +6475,89 @@ export const layer: Layer.Layer = position, canonicalPosition: position, })); - const historyItems = workMode === "deferred" ? deferWorkHistoryRows(indexed) : indexed; - const selection = yield* Effect.try({ - try: () => { - const policy = olderHistoryPagePolicy({ items: indexed, cursor, throughEntryId }); - const page = + const { olderPolicy, cursorPosition } = yield* Effect.try({ + 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) => + 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) { + // 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, + ); + if (cursorPosition >= 0) span.push(cursorPosition); + if (span.length === 0) break; + const runs = new Map( + // 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); + 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 IS 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/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 {