diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 0729b0a93f16..e00ee60c0b13 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -9,6 +9,7 @@ import { MessageId, type ModelSelection, type OrchestrationV2ProviderThread, + type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadShellSnapshot, NodeId, ProjectId, @@ -40,6 +41,7 @@ import { decodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY, + THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, } from "./threadHistoryPaging.ts"; const layerTest = Layer.mergeAll( @@ -5660,6 +5662,180 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("skips parent bodies once a fork's own turns fill its first page", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = DateTime.makeUnsafe("2026-10-01T00:00:00.000Z"); + const parentId = ThreadId.make("thread:filled-fork:parent"); + const forkId = ThreadId.make("thread:filled-fork:fork"); + const parentText = "parent prompt"; + const createThread = ( + threadId: ThreadId, + forkedFrom: { readonly threadId: ThreadId; readonly runId: RunId } | null, + ) => + store.apply({ + id: EventId.make(`event:${threadId}`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:filled-fork"), + title: "Filled fork", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: forkedFrom?.threadId ?? null, + relationshipToParent: forkedFrom === null ? null : "fork", + rootThreadId: parentId, + }, + forkedFrom: forkedFrom === null ? null : { type: "run", ...forkedFrom }, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + // Each turn is one run with one user prompt. + const addTurn = Effect.fnUntraced(function* (threadId: ThreadId, turn: number, text: string) { + const runId = RunId.make(`run:${threadId}:${turn}`); + const nodeId = NodeId.make(`node:${threadId}:${turn}`); + yield* store.apply({ + id: EventId.make(`event:${runId}`), + type: "run.created", + threadId, + runId, + nodeId, + driver, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: turn, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${runId}`), + rootNodeId: nodeId, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + yield* store.apply({ + id: EventId.make(`event:${runId}:prompt`), + type: "turn-item.updated", + threadId, + runId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:${runId}`), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: turn, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + messageId: MessageId.make(`message:${runId}`), + inputIntent: "turn_start", + createdBy: "user", + creationSource: "web", + text, + attachments: [], + }, + }); + }); + const firstPage = (projection: OrchestrationV2ThreadProjection) => { + const bounded = buildBoundedThreadProjection({ projection, snapshotSequence: 0 }); + return { + items: bounded.projection.visibleTurnItems, + historyCursor: bounded.historyCursor, + hasMoreHistory: bounded.hasMoreHistory, + }; + }; + // The initial window must be the tail of the full timeline, with the same + // positions, and bound to the same first page and cursor. + const readFirstPage = Effect.fnUntraced(function* (threadId: ThreadId) { + const full = (yield* store.getThreadSnapshot(threadId)).projection; + const { window, parsedParent } = yield* Effect.acquireUseRelease( + Effect.sync(() => vi.spyOn(JSON, "parse")), + (parse) => + store + .getThreadSnapshotWindow(threadId, { + rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }) + .pipe( + Effect.map(({ projection }) => ({ + window: projection, + parsedParent: parse.mock.calls.some(([text]) => text.includes(parentText)), + })), + ), + (parse) => Effect.sync(() => parse.mockRestore()), + ); + const visible = window.visibleTurnItems; + assert.deepEqual(visible, full.visibleTurnItems.slice(-visible.length)); + const page = firstPage(window); + assert.deepEqual(page, firstPage(full)); + return { page, parsedParent }; + }); + + yield* createThread(parentId, null); + for (const turn of [1, 2, 3]) yield* addTurn(parentId, turn, parentText); + // The parent kept going after the fork, so its third turn is not inherited. + yield* createThread(forkId, { threadId: parentId, runId: RunId.make(`run:${parentId}:2`) }); + for (let turn = 1; turn <= 9; turn += 1) yield* addTurn(forkId, turn, "fork prompt"); + + // Nine local turns leave room for the parent's last inherited turn, so + // the window still reads parent bodies. + const partial = yield* readFirstPage(forkId); + assert.include( + partial.page.items.map((row) => row.visibility), + "inherited", + ); + assert.isTrue(partial.parsedParent); + + // Ten local turns fill the page: parent rows only shift positions and + // the cursor, so their bodies are not read. + yield* addTurn(forkId, 10, "fork prompt"); + const filled = yield* readFirstPage(forkId); + assert.isNotNull(filled.page.historyCursor); + assert.isFalse(filled.parsedParent); + + // A fork of the fork inherits the grandparent's rows through its parent, + // and its parent's own ten turns fill a page too. Counting rows at either + // level would misplace the page, so a nested fork keeps the full read. + const nestedId = ThreadId.make("thread:filled-fork:nested"); + yield* createThread(nestedId, { threadId: forkId, runId: RunId.make(`run:${forkId}:10`) }); + for (let turn = 1; turn <= 10; turn += 1) yield* addTurn(nestedId, turn, "nested prompt"); + const nested = yield* readFirstPage(nestedId); + assert.isTrue(nested.parsedParent); + }), + ); + it.effect("a pull request watch keeps a finished thread working until it ends", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ad92e25c8516..5da46f4c978a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1307,8 +1307,9 @@ function withLocalVisibleTurnItems( function renumberVisibleTurnItems( rows: ReadonlyArray>, + offset = 0, ): Array { - return rows.map((row, position) => ({ ...row, position })); + return rows.map((row, position) => ({ ...row, position: position + offset })); } function makeForkMarkerTurnItem(input: { @@ -1396,16 +1397,24 @@ function visibleTurnItemsThroughRun(input: { function buildVisibleTurnItems(input: { readonly projection: OrchestrationV2ThreadProjection; readonly sourceProjection: OrchestrationV2ThreadProjection | null; + /** Inherited rows a window counted instead of reading; positions start after them. */ + readonly inheritedRowCount?: number; }): Array { const forkedFrom = input.projection.thread.forkedFrom; - if (forkedFrom?.type !== "run" || input.sourceProjection === null) { + if ( + forkedFrom?.type !== "run" || + (input.sourceProjection === null && input.inheritedRowCount === undefined) + ) { return localVisibleTurnItems(input.projection); } - const inherited = visibleTurnItemsThroughRun({ - sourceProjection: input.sourceProjection, - sourceRunId: forkedFrom.runId, - }); + const inherited = + input.sourceProjection === null + ? [] + : visibleTurnItemsThroughRun({ + sourceProjection: input.sourceProjection, + sourceRunId: forkedFrom.runId, + }); const markerItem = makeForkMarkerTurnItem({ targetProjection: input.projection, sourceThreadId: forkedFrom.threadId, @@ -1418,16 +1427,19 @@ function buildVisibleTurnItems(input: { item: row.item, })); - return renumberVisibleTurnItems([ - ...inherited, - { - visibility: "synthetic", - sourceThreadId: forkedFrom.threadId, - sourceItemId: markerItem.id, - item: markerItem, - }, - ...local, - ]); + return renumberVisibleTurnItems( + [ + ...inherited, + { + visibility: "synthetic", + sourceThreadId: forkedFrom.threadId, + sourceItemId: markerItem.id, + item: markerItem, + }, + ...local, + ], + input.inheritedRowCount ?? 0, + ); } /** @@ -2762,45 +2774,19 @@ export const layer: Layer.Layer = ), ); - const readCanonicalProjection = ( + // The rows a bounded window keeps for one thread: the newest user turns up + // to the limit (rows up to the limit when nothing starts a turn), the + // interrupt requests they depend on, and the latest row. Ends with the + // `retained` CTE so a read can fetch the payloads or only count the rows. + const retainedWindowRows = ( threadId: ThreadId, - window?: { + window: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly requiredRunId?: RunId | undefined; - readonly suppressLocal?: boolean | undefined; }, - fields?: ReadonlyArray, - filter?: ProjectionRecordFilter, - ) => - Effect.gen(function* () { - const threadRows = yield* sql` - SELECT payload_json - FROM orchestration_v2_projection_threads - WHERE thread_id = ${threadId} - LIMIT 1 - `; - const threadRow = threadRows[0]; - if (!threadRow) { - return yield* new ProjectionStoreThreadNotFoundError({ threadId }); - } - - const boundedTurnItemRows = - fields !== undefined && !fields.includes("turnItems") - ? [] - : window === undefined - ? yield* sql` - SELECT payload_json - FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} - ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} - ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} - ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} - ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} - ORDER BY ordinal ASC, turn_item_id ASC - ` - : yield* sql` + ) => sql` WITH eligible AS NOT MATERIALIZED ( SELECT item.rowid AS item_rowid, item.payload_json, item.ordinal, item.turn_item_id, item.run_id, item.node_id, item.type @@ -2965,6 +2951,48 @@ export const layer: Layer.Layer = ) AS latest ORDER BY ordinal ASC, turn_item_id ASC ) + `; + + const readCanonicalProjection = ( + threadId: ThreadId, + window?: { + readonly rowLimit: number; + readonly userTurnLimit?: number | undefined; + readonly anchorItemId?: TurnItemId | undefined; + readonly requiredRunId?: RunId | undefined; + readonly suppressLocal?: boolean | undefined; + }, + fields?: ReadonlyArray, + filter?: ProjectionRecordFilter, + ) => + Effect.gen(function* () { + const threadRows = yield* sql` + SELECT payload_json + FROM orchestration_v2_projection_threads + WHERE thread_id = ${threadId} + LIMIT 1 + `; + const threadRow = threadRows[0]; + if (!threadRow) { + return yield* new ProjectionStoreThreadNotFoundError({ threadId }); + } + + const boundedTurnItemRows = + fields !== undefined && !fields.includes("turnItems") + ? [] + : window === undefined + ? yield* sql` + SELECT payload_json + FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadId} + ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} + ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} + ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} + ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} + ORDER BY ordinal ASC, turn_item_id ASC + ` + : yield* sql` + ${retainedWindowRows(threadId, window)} -- CROSS JOIN keeps the sorted IDs as the outer loop, so SQLite -- skips sorting again once the payloads are attached. Carry the -- rowid through selection to avoid a second string-key index lookup. @@ -3557,6 +3585,49 @@ export const layer: Layer.Layer = }; }); + // A fork whose own turns fill its first page shows no inherited row + // there, but the inherited count still sets the page's positions and + // history cursor. Count the parent's retained rows instead of reading + // and decoding their bodies. The count mirrors visibleTurnItemsThroughRun + // for a parent with no inherited rows of its own: a parent that is itself + // a run fork returns NULL and keeps the full ancestry read. + if ( + window?.userTurnLimit !== undefined && + window.historyAnchor === undefined && + seenThreadIds.size === 0 && + sourceWindow !== undefined && + activeLocalTurnItems(projection).filter((row) => isThreadHistoryUserTurn(row.item)) + .length >= Math.max(window.userTurnLimit, THREAD_HISTORY_PAGE_POLICY.maxUserTurns) + ) { + const counted = yield* sql<{ readonly inherited_rows: number | null }>` + ${retainedWindowRows(forkedFrom.threadId, { + ...sourceWindow, + rowLimit: sourceWindow.suppressLocal ? 0 : sourceWindow.rowLimit, + })} + SELECT CASE + WHEN json_extract(source.payload_json, '$.forkedFrom.type') = 'run' THEN NULL + WHEN NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs + WHERE thread_id = source.thread_id AND run_id = ${forkedFrom.runId} + ) THEN 0 + ELSE (SELECT COUNT(*) FROM retained) + END AS inherited_rows + FROM orchestration_v2_projection_threads AS source + WHERE source.thread_id = ${forkedFrom.threadId} + `; + const inheritedRowCount = counted[0]?.inherited_rows ?? null; + if (inheritedRowCount !== null) { + return { + ...projection, + visibleTurnItems: buildVisibleTurnItems({ + projection, + sourceProjection: null, + inheritedRowCount, + }), + }; + } + } + const sourceProjection = yield* readProjection( forkedFrom.threadId, new Set([...seenThreadIds, threadId]),