From afee90dc00a42d5b6f2d4c4486dc9c16f7f6e81c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:19:19 +0000 Subject: [PATCH 1/9] test(server): cover viewport snapshots and deferred history support --- .../orchestration-v2/ProjectionStore.test.ts | 138 +++++++++++++++++- .../threadHistoryPaging.test.ts | 59 ++++++++ 2 files changed, 196 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 74377160a8204..2b56b0bc8bf4f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -41,6 +41,7 @@ import { decodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY, + THREAD_HISTORY_MAX_RAW_TURNS, } from "./threadHistoryPaging.ts"; const layerTest = Layer.mergeAll( @@ -1395,6 +1396,26 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ]; startIds.push(rows[1]!.turn_item_id); yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`; + yield* projectionStore.apply({ + id: EventId.make(`event:notification-turn-pages:checkpoint:${run}`), + type: "checkpoint.captured", + threadId, + occurredAt: now, + payload: { + id: CheckpointId.make(`checkpoint:notification-turn-pages:${run}`), + threadId, + scopeId: CheckpointScopeId.make(`scope:notification-turn-pages:${run}`), + runId: RunId.make(runId), + nodeId: NodeId.make(`node:notification-turn-pages:${run}`), + parentCheckpointId: null, + ordinalWithinScope: 1, + appRunOrdinal: run, + ref: CheckpointRef.make(`refs/t3/notification-turn-pages/${run}`), + status: "ready", + capturedAt: now, + files: [{ path: `run-${run}.ts`, kind: "modified", additions: run, deletions: 1 }], + }, + }); } const { statements, tracer } = traceSqlStatements(); @@ -1445,6 +1466,73 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { cursor = page.nextCursor; } assert.deepEqual(loaded, allIds); + + // One old user turn must not make notification-started runs an unlimited support cohort. + yield* sql`UPDATE orchestration_v2_projection_turn_items + SET type = 'notification', payload_json = json_set(payload_json, + '$.type', 'notification', '$.source', json('{"kind":"monitor"}'), + '$.outcome', 'completed', '$.summary', 'Wake') + WHERE thread_id = ${threadId} AND type = 'user_message' + AND run_id <> ${"run:notification-turn-pages:1"}`; + const viewport = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + deferHistoricalSupport: true, + }); + assert.isAtMost(viewport.projection.checkpoints.length, THREAD_HISTORY_MAX_RAW_TURNS + 2); + const boundedViewport = buildBoundedThreadProjection({ ...viewport, viewportWindow: true }); + const checkpoints = boundedViewport.projection.checkpoints; + assert.isAtMost(checkpoints.length, THREAD_HISTORY_MAX_RAW_TURNS + 2); + assert.isFalse(checkpoints.some((checkpoint) => checkpoint.appRunOrdinal === 1)); + const visibleRunIds = new Set( + boundedViewport.projection.visibleTurnItems.map((row) => row.item.runId), + ); + const deferred = checkpoints.filter((checkpoint) => !visibleRunIds.has(checkpoint.runId)); + assert.isAbove( + deferred.length, + 0, + "recent inactive turns remain available to Diff and Review", + ); + const loadedRunIds = new Set(viewport.projection.turnItems.map((item) => item.runId)); + assert.isTrue(deferred.some((checkpoint) => !loadedRunIds.has(checkpoint.runId))); + for (const checkpoint of deferred) { + assert.deepEqual(checkpoint.files, []); + assert.strictEqual(checkpoint.deferredFileCount, 1); + } + assert.deepEqual( + checkpoints.find((checkpoint) => checkpoint.appRunOrdinal === startedRuns)?.files, + [{ path: `run-${startedRuns}.ts`, kind: "modified", additions: startedRuns, deletions: 1 }], + ); + + assert.isNotNull(boundedViewport.historyCursor); + const history = yield* projectionStore.getThreadHistoryPage( + threadId, + boundedViewport.historyCursor!, + undefined, + false, + true, + ); + const historyRunIds = new Set(history.items.map((row) => row.item.runId)); + const recovered = deferred.filter((checkpoint) => historyRunIds.has(checkpoint.runId)); + assert.isAbove(recovered.length, 0); + for (const checkpoint of recovered) { + const hydrated = history.support?.checkpoints.find((entry) => entry.id === checkpoint.id); + assert.deepEqual(hydrated?.files, [ + { + path: `run-${checkpoint.appRunOrdinal}.ts`, + kind: "modified", + additions: checkpoint.appRunOrdinal, + deletions: 1, + }, + ]); + assert.isUndefined(hydrated?.deferredFileCount); + assert.isTrue(history.support?.runs.some((run) => run.id === checkpoint.runId)); + } + const legacyHistory = yield* projectionStore.getThreadHistoryPage( + threadId, + boundedViewport.historyCursor!, + ); + assert.deepEqual(legacyHistory.items, history.items); + assert.isUndefined(legacyHistory.support); }), ); @@ -1570,7 +1658,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { providerTurnId: null, nativeItemRef: null, parentItemId: null, - ordinal: index, + ordinal: index + 1, status: "completed", title: null, startedAt: now, @@ -1673,6 +1761,54 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); assert.lengthOf(sharedNodeSnapshot.projection.visibleTurnItems, 76); assert.deepEqual(sharedNodeSnapshot.projection.nodes, snapshot.projection.nodes); + + const promptId = TurnItemId.make("item:bounded-node-history:prompt"); + yield* sql`INSERT INTO orchestration_v2_projection_turn_items + SELECT ${promptId}, thread_id, run_id, ${rootNodeId}, provider_thread_id, + provider_turn_id, parent_item_id, 0, 'user_message', status, updated_at, + json_set(payload_json, '$.id', ${promptId}, '$.nodeId', ${rootNodeId}, '$.ordinal', 0, + '$.type', 'user_message', '$.messageId', 'prompt', '$.text', 'Start the long run', + '$.attachments', json('[]'), '$.inputIntent', 'turn_start', '$.createdBy', 'user', '$.creationSource', 'web') + FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${duplicateItemId}`; + const viewport = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + deferHistoricalSupport: true, + }); + assert.isAtMost(viewport.projection.visibleTurnItems.length, 78); + assert.strictEqual(viewport.projection.visibleTurnItems[0]?.sourceItemId, promptId); + const boundedViewport = buildBoundedThreadProjection({ ...viewport, viewportWindow: true }); + assert.strictEqual(boundedViewport.projection.visibleTurnItems[0]?.sourceItemId, promptId); + assert.isAtMost(boundedViewport.projection.visibleTurnItems.length, 76); + assert.isTrue(boundedViewport.hasMoreHistory); + assert.deepEqual( + boundedViewport.projection.runtimeRequests, + snapshot.projection.runtimeRequests, + ); + for (const nodeId of [rootNodeId, parentNodeId, liveNodeId, requestNodeId]) { + assert.isTrue(boundedViewport.projection.nodes.some((node) => node.id === nodeId)); + } + assert.isFalse( + boundedViewport.projection.nodes.some( + (node) => node.id === NodeId.make("node:bounded-node-history:1"), + ), + ); + const olderPage = yield* projectionStore.getThreadHistoryPage( + threadId, + boundedViewport.historyCursor!, + undefined, + false, + true, + ); + assert.isTrue( + olderPage.items.some( + (row) => row.item.nodeId === NodeId.make("node:bounded-node-history:1"), + ), + ); + assert.isTrue( + olderPage.support?.nodes.some( + (node) => node.id === NodeId.make("node:bounded-node-history:1"), + ), + ); }), ); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 0b58163bbdff1..f36c1479624d4 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -175,6 +175,65 @@ describe("threadHistoryPaging", () => { expect(item.payloadBudgetExceeded).toBe(false); }); + it("opens a long completed turn at its viewport and keeps its prompt and interrupt dependency", () => { + const promptRow = makeRow(0); + const prompt: OrchestrationV2ProjectedTurnItem = { + ...promptRow, + item: { + ...promptRow.item, + type: "user_message", + runId: RUN, + createdBy: "user", + creationSource: "web", + inputIntent: "turn_start", + messageId: MessageId.make("long-turn-prompt"), + text: "Do the long task", + attachments: [], + }, + }; + const rows = [ + prompt, + interruptRow(1, "run_interrupt_request"), + ...Array.from({ length: 300 }, (_, index) => { + const row = makeRow(index + 2); + return { ...row, item: { ...row.item, runId: RUN } }; + }), + interruptRow(302, "run_interrupt_result"), + ]; + const full = makeProjection(rows); + const projection = { + ...full, + runs: full.runs.map((run) => ({ ...run, id: RUN, ordinal: 1, status: "completed" as const })), + }; + const legacy = buildBoundedThreadProjection({ projection, snapshotSequence: 7 }); + expect(legacy.projection.visibleTurnItems).toEqual(rows); + expect(legacy.hasMoreHistory).toBe(false); + + const viewport = buildBoundedThreadProjection({ + projection, + snapshotSequence: 7, + viewportWindow: true, + }); + const recent = rows.slice(-THREAD_HISTORY_PAGE_POLICY.maxItems); + expect(viewport.projection.visibleTurnItems).toHaveLength(recent.length + 1); + expect(viewport.projection.visibleTurnItems.map((row) => row.sourceItemId)).toEqual( + [prompt, ...recent].map((row) => row.sourceItemId), + ); + expect(viewport.projection.turnItems).toContainEqual(rows[1]!.item); + expect(viewport.latestLocalTurnOrdinal).toBe(303); + expect(viewport.hasMoreHistory).toBe(true); + expect(decodeThreadHistoryCursor(viewport.historyCursor!).si).toBe(recent[0]!.sourceItemId); + const older = selectHistoryPageFromCursor({ + items: rows, + cursor: viewport.historyCursor!, + snapshotSequence: 7, + }); + expect([...older.items, ...recent].map((row) => row.sourceItemId)).toEqual( + rows.map((row) => row.sourceItemId), + ); + expect(older.hasMoreHistory).toBe(false); + }); + it("keeps background turns with user turns, with main's 150-turn fan-out ceiling", () => { const items = Array.from({ length: 161 }, (_, turn) => { const row = makeRow(turn * 2); From 74ddb90a5109abfb6089c0c65bc97d7e8d779af6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:28:58 +0000 Subject: [PATCH 2/9] perf(server): add opt-in viewport snapshots and history support --- .../src/orchestration-v2/Orchestrator.ts | 5 +- .../src/orchestration-v2/ProjectionStore.ts | 318 +++++++++++++++--- .../ThreadManagementService.ts | 10 +- .../src/orchestration-v2/ThreadStream.ts | 2 + apps/server/src/orchestration-v2/http.ts | 19 +- .../orchestration-v2/threadHistoryPaging.ts | 68 +++- apps/server/src/ws.ts | 7 +- packages/contracts/src/environmentHttp.ts | 2 + packages/contracts/src/orchestrationV2.ts | 11 + 9 files changed, 388 insertions(+), 54 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 29ba67bdf6efe..5b92e944363d3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -323,6 +323,7 @@ export interface OrchestratorV2Shape { cursor: string, throughEntryId?: string, conversationOnly?: boolean, + includeSupport?: boolean, ) => Effect.Effect; readonly getTimelinePage: ( threadId: ThreadId, @@ -11595,9 +11596,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (cause) => new OrchestratorProjectionError({ threadId: input.threadId, cause }), ), ), - getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly, includeSupport) => projectionStore - .getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly) + .getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly, includeSupport) .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))), getTimelinePage: (threadId, options) => projectionStore diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa950..f3d7b7038e5d5 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -327,6 +327,40 @@ const isSettledLeafNode = (node: ProjectionNode, runs: ReadonlyArray run.id === node.runId && activeRunStatuses.has(run.status)); +function historySupportForItems( + projection: OrchestrationV2ThreadProjection, + items: ReadonlyArray, +) { + const runIds = new Set(items.map(({ item }) => item.runId)); + const runs = projection.runs.filter((run) => runIds.has(run.id)); + const attempts = projection.attempts.filter((attempt) => runIds.has(attempt.runId)); + const directNodeIds = new Set(items.map(({ item }) => item.nodeId)); + const nodeIds = new Set([ + ...directNodeIds, + ...runs.map((run) => run.rootNodeId), + ...attempts.map((attempt) => attempt.rootNodeId), + ]); + const byId = new Map(projection.nodes.map((node) => [node.id, node])); + for (const id of nodeIds) { + const parentId = id === null ? undefined : byId.get(id)?.parentNodeId; + if (parentId != null) nodeIds.add(parentId); + } + const checkpointIds = new Set( + items.flatMap(({ item }) => ("checkpointId" in item ? [item.checkpointId] : [])), + ); + return { + nodes: projection.nodes.filter((node) => nodeIds.has(node.id)), + checkpoints: projection.checkpoints.filter( + (checkpoint) => + (checkpoint.runId !== null && runIds.has(checkpoint.runId)) || + (checkpoint.nodeId !== null && directNodeIds.has(checkpoint.nodeId)) || + checkpointIds.has(checkpoint.id), + ), + runs, + attempts, + }; +} + export type ProjectionRecordField = Exclude< keyof OrchestrationV2ThreadProjection, "thread" | "updatedAt" | "visibleTurnItems" @@ -369,6 +403,7 @@ export interface ProjectionStoreV2Shape { cursor: string, throughEntryId?: string, conversationOnly?: boolean, + includeSupport?: boolean, ) => Effect.Effect; readonly getTimelinePage: ( threadId: ThreadId, @@ -533,6 +568,7 @@ export interface ProjectionStoreV2Shape { options: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; + readonly deferHistoricalSupport?: boolean | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly anchorThreadId?: ThreadId | undefined; readonly requiredRunId?: RunId | undefined; @@ -2735,6 +2771,7 @@ export const layer: Layer.Layer = window?: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; + readonly deferHistoricalSupport?: boolean | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly requiredRunId?: RunId | undefined; readonly suppressLocal?: boolean | undefined; @@ -2917,12 +2954,57 @@ export const layer: Layer.Layer = UNION SELECT request.rowid AS item_rowid, request.ordinal, request.turn_item_id FROM orchestration_v2_projection_turn_items AS request - WHERE request.run_id IN ( + WHERE request.thread_id = ${threadId} + AND request.run_id IN ( SELECT run_id FROM selected WHERE type = 'run_interrupt_result' AND run_id IS NOT NULL + UNION + SELECT run_id FROM orchestration_v2_projection_runs + WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 + AND thread_id = ${threadId} + AND status IN ('queued','preparing','starting','running','waiting') ) AND request.type = 'run_interrupt_request' UNION + SELECT prompt.rowid, prompt.ordinal, prompt.turn_item_id + FROM orchestration_v2_projection_turn_items AS prompt + WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 + AND prompt.thread_id = ${threadId} + AND prompt.run_id IN ( + SELECT run_id FROM ( + SELECT run_id FROM selected ORDER BY ordinal ASC, turn_item_id ASC LIMIT 1 + ) + UNION + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued','preparing','starting','running','waiting') + ) + AND prompt.rowid = ( + SELECT first_input.rowid FROM orchestration_v2_projection_turn_items first_input + WHERE first_input.run_id = prompt.run_id + AND (first_input.type = 'notification' OR first_input.type = 'user_message' + AND json_extract(first_input.payload_json, '$.inputIntent') IN ('turn_start','queued_turn')) + ORDER BY first_input.ordinal LIMIT 1 + ) + UNION + SELECT control.rowid, control.ordinal, control.turn_item_id + FROM orchestration_v2_projection_turn_items AS control + WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 + AND control.thread_id = ${threadId} + AND ( + (control.status IN ('pending','starting','running','waiting') + AND control.run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued','preparing','starting','running','waiting') + )) + OR (control.type IN ('approval_request','user_input_request') + AND json_extract(control.payload_json, '$.requestId') IN ( + SELECT runtime_request_id FROM orchestration_v2_projection_runtime_requests + WHERE thread_id = ${threadId} AND status IN ('pending','waiting') + )) + ) + UNION SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id FROM ( SELECT rowid AS item_rowid, ordinal, turn_item_id @@ -3010,6 +3092,16 @@ export const layer: Layer.Layer = const cohortMessageIds = cohortJson("messageId"); const cohortPlanIds = cohortJson("planId"); const cohortCheckpointIds = cohortJson("checkpointId"); + // Node ancestry is needed for controls, but must not hydrate historical file lists. + const checkpointNodeIds = window?.deferHistoricalSupport + ? cohortJson("nodeId") + : cohortNodeIds; + const checkpointRunIds = window?.deferHistoricalSupport + ? sql`SELECT value FROM json_each(${cohortRunIds}) + UNION SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued','preparing','starting','running','waiting')` + : sql`SELECT value FROM json_each(${cohortRunIds})`; const cohortHandoffIds = cohortJson("contextHandoffId"); const [ @@ -3274,12 +3366,26 @@ export const layer: Layer.Layer = ORDER BY scope_id ASC, ordinal_within_scope ASC ` : sql` - SELECT payload_json FROM orchestration_v2_projection_checkpoints + SELECT CASE + WHEN status IN ('pending','capturing') + OR checkpoint_id IN (SELECT value FROM json_each(${cohortCheckpointIds})) + OR run_id IN (${checkpointRunIds}) + OR node_id IN (SELECT value FROM json_each(${checkpointNodeIds})) + THEN payload_json + ELSE json_set(payload_json, '$.files', json('[]'), + '$.deferredFileCount', json_array_length(payload_json, '$.files')) + END AS payload_json + FROM orchestration_v2_projection_checkpoints WHERE thread_id = ${threadId} AND (status IN ('pending','capturing') OR checkpoint_id IN (SELECT value FROM json_each(${cohortCheckpointIds})) - OR run_id IN (SELECT value FROM json_each(${cohortRunIds})) - OR node_id IN (SELECT value FROM json_each(${cohortNodeIds}))) + OR run_id IN (${checkpointRunIds}) + OR node_id IN (SELECT value FROM json_each(${checkpointNodeIds})) + OR (${window.deferHistoricalSupport ? 1 : 0} = 1 AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} AND status <> 'rolled_back' + ORDER BY ordinal DESC LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2} + ))) ORDER BY scope_id ASC, ordinal_within_scope ASC `, fields !== undefined && !fields.includes("contextHandoffs") @@ -3400,6 +3506,7 @@ export const layer: Layer.Layer = window?: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; + readonly deferHistoricalSupport?: boolean | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly requiredRunId?: RunId | undefined; readonly suppressLocal?: boolean | undefined; @@ -3514,6 +3621,7 @@ export const layer: Layer.Layer = return { rowLimit: window.rowLimit, userTurnLimit: window.userTurnLimit, + deferHistoricalSupport: window.deferHistoricalSupport, requiredRunId: forkedFrom.runId, suppressLocal: anchor === undefined || anchorIsInDescendant, ...(anchor === undefined || anchorIsInDescendant @@ -5055,11 +5163,65 @@ export const layer: Layer.Layer = readTimelineIndex(threadId, new Set()).pipe(Effect.map((index) => index.visible)), ); + const readHistorySupport = Effect.fnUntraced(function* ( + threadId: ThreadId, + items: ReadonlyArray, + ) { + const runIds = encodeIdList([ + ...new Set(items.flatMap(({ item }) => (item.runId === null ? [] : [item.runId]))), + ]); + const nodeIds = encodeIdList([ + ...new Set(items.flatMap(({ item }) => (item.nodeId === null ? [] : [item.nodeId]))), + ]); + const checkpointIds = encodeIdList( + items.flatMap(({ item }) => ("checkpointId" in item ? [item.checkpointId] : [])), + ); + const sourceThreadIds = encodeIdList([...new Set(items.map((row) => row.sourceThreadId))]); + const [nodes, checkpoints, runs, attempts] = yield* Effect.all([ + sql` + WITH RECURSIVE retained(node_id) AS ( + SELECT value FROM json_each(${nodeIds}) + UNION + SELECT json_extract(payload_json, '$.rootNodeId') FROM orchestration_v2_projection_runs + WHERE run_id IN (SELECT value FROM json_each(${runIds})) + UNION + SELECT root_node_id FROM orchestration_v2_projection_run_attempts + WHERE run_id IN (SELECT value FROM json_each(${runIds})) + UNION + SELECT node.parent_node_id FROM orchestration_v2_projection_nodes node + JOIN retained ON retained.node_id = node.node_id + WHERE node.parent_node_id IS NOT NULL + ) + SELECT node.payload_json FROM retained + CROSS JOIN orchestration_v2_projection_nodes node ON node.node_id = retained.node_id + ORDER BY COALESCE(node.started_at, ''), node.node_id + `.pipe(Effect.flatMap(decodeRows(decodeNodePayload, threadId))), + sql` + SELECT payload_json FROM orchestration_v2_projection_checkpoints + WHERE thread_id IN (SELECT value FROM json_each(${sourceThreadIds})) + AND (run_id IN (SELECT value FROM json_each(${runIds})) + OR node_id IN (SELECT value FROM json_each(${nodeIds})) + OR checkpoint_id IN (SELECT value FROM json_each(${checkpointIds}))) + ORDER BY scope_id, ordinal_within_scope + `.pipe(Effect.flatMap(decodeRows(decodeCheckpointPayload, threadId))), + sql` + SELECT payload_json FROM orchestration_v2_projection_runs + WHERE run_id IN (SELECT value FROM json_each(${runIds})) ORDER BY ordinal + `.pipe(Effect.flatMap(decodeRows(decodeRunPayload, threadId))), + sql` + SELECT payload_json FROM orchestration_v2_projection_run_attempts + WHERE run_id IN (SELECT value FROM json_each(${runIds})) ORDER BY run_id, attempt_ordinal + `.pipe(Effect.flatMap(decodeRows(decodeRunAttemptPayload, threadId))), + ]); + return { nodes, checkpoints, runs, attempts }; + }); + const getThreadHistoryPage: ProjectionStoreV2Shape["getThreadHistoryPage"] = ( threadId, cursor, throughEntryId, conversationOnly, + includeSupport, ) => sql .withTransaction( @@ -5130,29 +5292,31 @@ export const layer: Layer.Layer = }; }), ); - if (hasUserTurns) - return { - ...page, - snapshotSequence, - items: items.map((row, position) => ({ ...row, position })), - }; - const bounded = selectRecentTimelineWindow({ - items, - snapshotSequence, - policy: { - ...THREAD_HISTORY_PAGE_POLICY, - maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, - maxItems: - THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2), - }, - }); + const bounded = hasUserTurns + ? undefined + : selectRecentTimelineWindow({ + items, + snapshotSequence, + policy: { + ...THREAD_HISTORY_PAGE_POLICY, + maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, + maxItems: + THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2), + }, + }); + const selectedItems = bounded?.items ?? items; return { snapshotSequence, - items: conversationOnly - ? bounded.items.filter((row) => isConversationHistoryItem(row.item)) - : bounded.items, - nextCursor: bounded.hasMoreHistory ? bounded.nextCursor : page.nextCursor, - hasMoreHistory: bounded.hasMoreHistory || page.hasMoreHistory, + items: hasUserTurns + ? selectedItems.map((row, position) => ({ ...row, position })) + : conversationOnly + ? selectedItems.filter((row) => isConversationHistoryItem(row.item)) + : selectedItems, + nextCursor: bounded?.hasMoreHistory ? bounded.nextCursor : page.nextCursor, + hasMoreHistory: bounded?.hasMoreHistory === true || page.hasMoreHistory, + ...(includeSupport + ? { support: yield* readHistorySupport(threadId, selectedItems) } + : {}), }; }), ) @@ -5246,7 +5410,8 @@ export const layer: Layer.Layer = }; const projection = yield* readProjection(threadId, new Set(), { rowLimit: options.rowLimit, - userTurnLimit: options.userTurnLimit, + userTurnLimit: options.deferHistoricalSupport ? undefined : options.userTurnLimit, + deferHistoricalSupport: options.deferHistoricalSupport, ...(historyAnchor?.threadId === threadId ? { anchorItemId: historyAnchor.itemId } : {}), @@ -6747,15 +6912,15 @@ export const layerMemory: Layer.Layer = Layer.effect( ), ), ), - getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly, includeSupport) => service.getThreadSnapshot(threadId).pipe( Effect.flatMap(({ projection, snapshotSequence }) => Effect.gen(function* () { if (projection.thread.deletedAt !== null) return yield* new ProjectionStoreThreadNotFoundError({ threadId }); - return yield* Effect.try({ - try: () => { - const page = selectHistoryPageFromCursor({ + const page = yield* Effect.try({ + try: () => + selectHistoryPageFromCursor({ items: projection.visibleTurnItems.map((row) => ({ ...row, item: projectTurnItemForWire(row.item), @@ -6763,17 +6928,40 @@ export const layerMemory: Layer.Layer = Layer.effect( cursor, throughEntryId, snapshotSequence, - }); - return { - ...page, - snapshotSequence, - items: conversationOnly - ? page.items.filter((row) => isConversationHistoryItem(row.item)) - : page.items, - }; - }, + }), catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), }); + const supports = includeSupport + ? yield* Effect.forEach( + [...new Set(page.items.map((row) => row.sourceThreadId))], + (sourceId) => + service.getThreadSnapshot(sourceId).pipe( + Effect.map(({ projection }) => + historySupportForItems( + projection, + page.items.filter((row) => row.sourceThreadId === sourceId), + ), + ), + ), + ) + : undefined; + return { + ...page, + snapshotSequence, + items: conversationOnly + ? page.items.filter((row) => isConversationHistoryItem(row.item)) + : page.items, + ...(supports === undefined + ? {} + : { + support: { + nodes: supports.flatMap((support) => support.nodes), + checkpoints: supports.flatMap((support) => support.checkpoints), + runs: supports.flatMap((support) => support.runs), + attempts: supports.flatMap((support) => support.attempts), + }, + }), + }; }), ), ), @@ -6806,7 +6994,7 @@ export const layerMemory: Layer.Layer = Layer.effect( const candidates = snapshot.projection.visibleTurnItems.slice(0, anchorIndex); const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item)); const turnAnchors = - options.userTurnLimit === undefined + options.userTurnLimit === undefined || options.deferHistoricalSupport ? [] : candidates.flatMap((_row, index) => (turnStarts[index] ? [index] : [])); const rawStart = turnAnchors.at(-(THREAD_HISTORY_MAX_RAW_TURNS + 2)) ?? 0; @@ -6820,10 +7008,56 @@ export const layerMemory: Layer.Layer = Layer.effect( ? rawStart : anchors.at(-anchorLimit)! : Math.max(0, anchorIndex - options.rowLimit); - const visibleTurnItems = candidates.slice(start); + const windowItems = candidates.slice(start); + const firstRunId = windowItems[0]?.item.runId; + const prompt = + options.deferHistoricalSupport && firstRunId != null + ? candidates.find( + (row, index) => row.item.runId === firstRunId && turnStarts[index], + ) + : undefined; + const visibleTurnItems = + prompt !== undefined && !windowItems.includes(prompt) + ? [prompt, ...windowItems] + : windowItems; + const recentRunIds = new Set( + snapshot.projection.runs + .filter((run) => run.status !== "rolled_back") + .slice(-(THREAD_HISTORY_MAX_RAW_TURNS + 2)) + .map((run) => run.id), + ); + const fullRunIds = new Set([ + ...visibleTurnItems.map(({ item }) => item.runId), + ...snapshot.projection.runs + .filter((run) => activeRunStatuses.has(run.status)) + .map((run) => run.id), + ]); return { ...snapshot, - projection: { ...snapshot.projection, visibleTurnItems }, + projection: { + ...snapshot.projection, + visibleTurnItems, + ...(options.deferHistoricalSupport + ? { + checkpoints: snapshot.projection.checkpoints + .filter( + (checkpoint) => + checkpoint.runId !== null && + (recentRunIds.has(checkpoint.runId) || + fullRunIds.has(checkpoint.runId)), + ) + .map((checkpoint) => + fullRunIds.has(checkpoint.runId) + ? checkpoint + : { + ...checkpoint, + files: [], + deferredFileCount: checkpoint.files.length, + }, + ), + } + : {}), + }, }; }), ), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d83dd9578c279..7fca702933e7a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1131,10 +1131,16 @@ const make = Effect.gen(function* () { Effect.andThen(orchestrator.searchThread(input)), ), dispatch, - getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly, includeSupport) => ensureProjectionTranscript(threadId).pipe( Effect.andThen( - orchestrator.getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly), + orchestrator.getThreadHistoryPage( + threadId, + cursor, + throughEntryId, + conversationOnly, + includeSupport, + ), ), ), getTimelinePage: (threadId, options) => diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index 4ef21edd3b3c7..7cf6147db04ea 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -61,10 +61,12 @@ export function buildBoundedThreadStreamSnapshot(input: { readonly snapshotSequence: number; readonly projection: OrchestrationV2ThreadProjection; readonly compactTurnItems?: boolean; + readonly viewportWindow?: boolean; }): ThreadSnapshotStreamItem { const bounded = buildBoundedThreadProjection({ snapshotSequence: input.snapshotSequence, projection: projectThreadProjectionForWire(input.projection), + viewportWindow: input.viewportWindow === true, }); return { kind: "snapshot", diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index cc8e987880df8..396a18e38368e 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -120,12 +120,15 @@ export const layer = HttpApiBuilder.group( anchorThreadId?: Parameters< typeof threadManagement.getThreadSnapshotWindow >[1]["anchorThreadId"], + viewportWindow = false, ) { return yield* threadManagement .getThreadSnapshotWindow(threadId, { rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, - userTurnLimit: - anchorItemId === undefined + deferHistoricalSupport: viewportWindow, + userTurnLimit: viewportWindow + ? undefined + : anchorItemId === undefined ? THREAD_HISTORY_PAGE_POLICY.maxUserTurns : OLDER_THREAD_USER_TURN_LIMIT, ...(anchorItemId === undefined ? {} : { anchorItemId }), @@ -207,12 +210,17 @@ export const layer = HttpApiBuilder.group( Effect.fn("environment.orchestration.threadBoundedSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); - const snapshot = yield* loadThreadSnapshotWindow(args.params.threadId).pipe( - traceLocalHandlerWork, - ); + const viewportWindow = args.query.viewportWindow === "1"; + const snapshot = yield* loadThreadSnapshotWindow( + args.params.threadId, + undefined, + undefined, + viewportWindow, + ).pipe(traceLocalHandlerWork); const bounded = buildBoundedThreadProjection({ projection: snapshot.projection, snapshotSequence: snapshot.snapshotSequence, + viewportWindow, }); return { snapshotSequence: snapshot.snapshotSequence, @@ -241,6 +249,7 @@ export const layer = HttpApiBuilder.group( args.query.cursor, args.query.throughEntryId, args.query.view === "conversation", + args.query.includeSupport === "1", ) .pipe( traceLocalHandlerWork, diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index ee675856477cc..b2727b5a47dae 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -458,11 +458,35 @@ export function buildBoundedThreadProjection(input: { readonly projection: OrchestrationV2ThreadProjection; readonly snapshotSequence: number; readonly policy?: ThreadHistoryPagePolicy | undefined; + readonly viewportWindow?: boolean; }): BoundedProjectionResult { const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY; const threadId = input.projection.thread.id; + const recentRows = input.projection.visibleTurnItems.slice(-policy.maxItems); + const retainedRunIds = new Set(recentRows.map(({ item }) => item.runId)); + for (const run of input.projection.runs) { + if (["queued", "preparing", "starting", "running", "waiting"].includes(run.status)) { + retainedRunIds.add(run.id); + } + } + const retainedNodeIds = new Set(recentRows.map(({ item }) => item.nodeId)); + const retainedCheckpointIds = new Set( + recentRows.flatMap(({ item }) => ("checkpointId" in item ? [item.checkpointId] : [])), + ); const controlProjection = { ...input.projection, + ...(input.viewportWindow + ? { + checkpoints: input.projection.checkpoints.map((checkpoint) => + (checkpoint.runId !== null && retainedRunIds.has(checkpoint.runId)) || + (checkpoint.nodeId !== null && retainedNodeIds.has(checkpoint.nodeId)) || + retainedCheckpointIds.has(checkpoint.id) || + checkpoint.deferredFileCount !== undefined + ? checkpoint + : { ...checkpoint, files: [], deferredFileCount: checkpoint.files.length }, + ), + } + : {}), plans: input.projection.plans.filter((plan) => plan.status === "active"), contextHandoffs: input.projection.contextHandoffs.filter( (handoff) => handoff.status === "pending" || handoff.status === "ready", @@ -492,7 +516,7 @@ export function buildBoundedThreadProjection(input: { policy.maxEncodedBytes - dependencyReserve - controlBytes - 1_024, ); const windowPolicy: ThreadHistoryPagePolicy = { - maxUserTurns: policy.maxUserTurns, + maxUserTurns: input.viewportWindow ? undefined : policy.maxUserTurns, maxItems: policy.maxItems, maxEncodedBytes: windowBudget, }; @@ -503,7 +527,21 @@ export function buildBoundedThreadProjection(input: { policy: windowPolicy, rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId), }); - const visibleTurnItems = renumberPositions(window.items); + const oldest = window.items[0]; + const prompt = + input.viewportWindow && oldest?.item.runId != null + ? controlProjection.visibleTurnItems.find( + (row) => + row.item.runId === oldest.item.runId && + (isThreadHistoryTurnStart(row.item) || row.item.type === "notification"), + ) + : undefined; + // The cursor stays at the row window so paging fills the gap below its prompt. + const visibleTurnItems = renumberPositions( + prompt === undefined || window.items.includes(prompt) + ? window.items + : [prompt, ...window.items], + ); const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems); const dependencyTurnItems = retainedInterruptRequestTurnItems( controlProjection, @@ -516,6 +554,32 @@ export function buildBoundedThreadProjection(input: { for (const item of dependencyTurnItems) { turnItemById.set(String(item.id), item); } + if (input.viewportWindow) { + const activeRunIds = new Set( + controlProjection.runs + .filter((run) => + ["queued", "preparing", "starting", "running", "waiting"].includes(run.status), + ) + .map((run) => run.id), + ); + const pendingRequestIds = new Set( + controlProjection.runtimeRequests + .filter((request) => request.status === "pending") + .map((request) => request.id), + ); + for (const item of controlProjection.turnItems) { + if ( + (item.runId !== null && + activeRunIds.has(item.runId) && + (item.type === "user_message" || + item.type === "notification" || + ["pending", "starting", "running", "waiting"].includes(item.status))) || + ("requestId" in item && pendingRequestIds.has(item.requestId)) + ) { + turnItemById.set(String(item.id), item); + } + } + } const turnItems = [...turnItemById.values()]; const visiblePlanIds = new Set( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index ade826e53f38b..a80e02f4665b5 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -741,6 +741,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs readonly requestCompletionMarker?: boolean; readonly acceptBoundedSnapshot?: boolean; readonly acceptCompactTurnItems?: boolean; + readonly acceptViewportWindow?: boolean; }) { const threadManagement = yield* ThreadManagementService.ThreadManagementService; const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; @@ -823,7 +824,10 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs useBoundedSnapshot ? threadManagement.getThreadSnapshotWindow(input.threadId, { rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, - userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + userTurnLimit: input.acceptViewportWindow + ? undefined + : THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + deferHistoricalSupport: input.acceptViewportWindow === true, }) : threadManagement.getThreadSnapshot(input.threadId) ).pipe( @@ -841,6 +845,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ? buildBoundedThreadStreamSnapshot({ ...snapshot, compactTurnItems: input.acceptCompactTurnItems === true, + viewportWindow: input.acceptViewportWindow === true, }) : { kind: "snapshot" as const, diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 31b69997f43cf..dc3ad84d68494 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -574,10 +574,12 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ }); const EnvironmentOrchestrationThreadBoundedSnapshotQuery = Schema.Struct({ + viewportWindow: Schema.optionalKey(Schema.String), compactTurnItems: Schema.optionalKey(Schema.String), }); const EnvironmentOrchestrationThreadHistoryQuery = Schema.Struct({ + includeSupport: Schema.optionalKey(Schema.String), cursor: TrimmedNonEmptyString, throughEntryId: Schema.optional(TrimmedNonEmptyString), view: Schema.optional(Schema.Literals(["conversation", "activity"])), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..86096456f000f 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1184,6 +1184,8 @@ export const OrchestrationV2Checkpoint = Schema.Struct({ ref: CheckpointRef, status: Schema.Literals(["ready", "missing", "error", "stale"]), files: Schema.Array(OrchestrationV2CheckpointFileSummary), + /** Historical file lists hydrate with their history page. */ + deferredFileCount: Schema.optional(NonNegativeInt), capturedAt: Schema.DateTimeUtc, }); export type OrchestrationV2Checkpoint = typeof OrchestrationV2Checkpoint.Type; @@ -3190,6 +3192,7 @@ export const OrchestrationV2SubscribeThreadInput = Schema.Struct({ requestCompletionMarker: Schema.optionalKey(Schema.Boolean), acceptBoundedSnapshot: Schema.optionalKey(Schema.Boolean), acceptCompactTurnItems: Schema.optionalKey(Schema.Boolean), + acceptViewportWindow: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationV2SubscribeThreadInput = typeof OrchestrationV2SubscribeThreadInput.Type; @@ -3237,6 +3240,14 @@ export const OrchestrationV2ThreadHistoryPage = Schema.Struct({ ), nextCursor: Schema.NullOr(TrimmedNonEmptyString), hasMoreHistory: Schema.Boolean, + support: Schema.optionalKey( + Schema.Struct({ + nodes: Schema.Array(OrchestrationV2ExecutionNode), + checkpoints: Schema.Array(OrchestrationV2Checkpoint), + runs: Schema.Array(OrchestrationV2Run), + attempts: Schema.Array(OrchestrationV2RunAttempt), + }), + ), }); export type OrchestrationV2ThreadHistoryPage = typeof OrchestrationV2ThreadHistoryPage.Type; From 58f2645a452a923cb9ce31d9754fe51b9fbdafd7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:33:05 +0000 Subject: [PATCH 3/9] test(server): narrow viewport fixture types --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 1 + apps/server/src/orchestration-v2/threadHistoryPaging.test.ts | 1 + 2 files changed, 2 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 2b56b0bc8bf4f..d6a70cef8572a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1515,6 +1515,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const recovered = deferred.filter((checkpoint) => historyRunIds.has(checkpoint.runId)); assert.isAbove(recovered.length, 0); for (const checkpoint of recovered) { + if (checkpoint.appRunOrdinal === null) throw new Error("Expected a run ordinal"); const hydrated = history.support?.checkpoints.find((entry) => entry.id === checkpoint.id); assert.deepEqual(hydrated?.files, [ { diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index f36c1479624d4..c8979fd3bedb9 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -177,6 +177,7 @@ describe("threadHistoryPaging", () => { it("opens a long completed turn at its viewport and keeps its prompt and interrupt dependency", () => { const promptRow = makeRow(0); + if (promptRow.item.type !== "command_execution") throw new Error("Expected command fixture"); const prompt: OrchestrationV2ProjectedTurnItem = { ...promptRow, item: { From c0912c71f09b6bd678c8e597b5362bf77809a7c3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:00:57 +0000 Subject: [PATCH 4/9] test(server): cover fork history support and viewport prompts --- .../orchestration-v2/ProjectionStore.test.ts | 56 +++++++ .../src/orchestration-v2/ThreadFind.test.ts | 154 +++++++++++++++++- 2 files changed, 201 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index d6a70cef8572a..7c0f9ad82b492 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1810,6 +1810,62 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { (node) => node.id === NodeId.make("node:bounded-node-history:1"), ), ); + + const queuedRunId = RunId.make("run:bounded-node-history:queued"); + const queuedPromptId = TurnItemId.make("item:bounded-node-history:queued-prompt"); + yield* sql`INSERT INTO orchestration_v2_projection_runs ( + run_id, thread_id, ordinal, provider, provider_instance_id, provider_thread_id, + status, requested_at, completed_at, payload_json + ) + SELECT ${queuedRunId}, thread_id, ordinal + 1, provider, provider_instance_id, provider_thread_id, + 'completed', requested_at, ${DateTime.formatIso(now)}, + json_set(payload_json, '$.id', ${queuedRunId}, '$.ordinal', ordinal + 1, + '$.status', 'completed', '$.completedAt', ${DateTime.formatIso(now)}, + '$.rootNodeId', NULL, '$.userMessageId', ${queuedPromptId}) + FROM orchestration_v2_projection_runs WHERE run_id = ${runId}`; + // A queued prompt precedes the previous run's final activity in thread ordinal order. + yield* sql`INSERT INTO orchestration_v2_projection_turn_items + SELECT ${queuedPromptId}, thread_id, ${queuedRunId}, NULL, provider_thread_id, + provider_turn_id, parent_item_id, 1002, type, status, updated_at, + json_set(payload_json, '$.id', ${queuedPromptId}, '$.messageId', ${queuedPromptId}, + '$.runId', ${queuedRunId}, '$.nodeId', NULL, '$.ordinal', 1002, + '$.inputIntent', 'queued_turn', '$.text', 'Start the queued run') + FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${promptId}`; + for (let index = 0; index < 77; index++) { + const id = TurnItemId.make(`item:bounded-node-history:interleaved:${index}`); + const ownerRunId = index < 2 ? runId : queuedRunId; + yield* sql`INSERT INTO orchestration_v2_projection_turn_items + SELECT ${id}, thread_id, ${ownerRunId}, NULL, provider_thread_id, + provider_turn_id, parent_item_id, ${1003 + index}, type, status, updated_at, + json_set(payload_json, '$.id', ${id}, '$.runId', ${ownerRunId}, + '$.nodeId', NULL, '$.ordinal', ${1003 + index}) + FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${duplicateItemId}`; + } + yield* sql`UPDATE orchestration_v2_projection_runs + SET status = 'completed', payload_json = json_set(payload_json, + '$.status', 'completed', '$.completedAt', ${DateTime.formatIso(now)}) + WHERE run_id = ${runId}`; + const interleaved = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + deferHistoricalSupport: true, + }); + assert.deepEqual( + interleaved.projection.visibleTurnItems.slice(-77).map((row) => row.item.runId), + [runId, runId, ...Array.from({ length: 75 }, () => queuedRunId)], + ); + assert.strictEqual( + interleaved.projection.runs.find((run) => run.id === queuedRunId)?.status, + "completed", + ); + const boundedInterleaved = buildBoundedThreadProjection({ + ...interleaved, + viewportWindow: true, + }); + assert.strictEqual( + boundedInterleaved.projection.visibleTurnItems[0]?.sourceItemId, + queuedPromptId, + ); + assert.lengthOf(boundedInterleaved.projection.visibleTurnItems, 76); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index 0121227f6e86a..4856bdb9bf114 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -1,7 +1,12 @@ import { assert, it, describe } from "@effect/vitest"; import { + CheckpointId, + CheckpointRef, + CheckpointScopeId, EventId, MessageId, + NodeId, + RunAttemptId, PlanId, RunId, RuntimeRequestId, @@ -12,6 +17,7 @@ import { type OrchestrationV2ThreadHistoryPage, ProjectId, ProviderInstanceId, + ProviderThreadId, ThreadId, type OrchestrationV2DomainEvent, } from "@t3tools/contracts"; @@ -695,9 +701,12 @@ describe("V2 thread find", () => { }).pipe(Effect.provide(layerTest)), ); - it.effect( - "searches inherited fork history through the fork boundary, using source item identities", - () => + it.effect.each([ + { storeName: "SQL", projectionLayer: ProjectionStore.layer }, + { storeName: "memory", projectionLayer: ProjectionStore.layerMemory }, + ])( + "$storeName searches inherited fork history without importing parent support", + ({ projectionLayer }) => Effect.gen(function* () { const projection = yield* setup; const parent = ThreadId.make("thread:parent"); @@ -708,8 +717,96 @@ describe("V2 thread find", () => { run(parent, parentRun), run(parent, laterRun, 2), ]); + for (const [owner, ownerRun] of [ + [parent, parentRun], + [threadId, runId], + ] as const) { + const nodeId = NodeId.make(`node:${owner}`); + const attemptId = RunAttemptId.make(`attempt:${owner}`); + const checkpointId = CheckpointId.make(`checkpoint:${owner}`); + const createdRun = run(owner, ownerRun); + yield* commit([ + { + ...createdRun, + id: EventId.make(`support-run:${owner}`), + type: "run.updated", + payload: { + ...createdRun.payload, + rootNodeId: nodeId, + activeAttemptId: attemptId, + checkpointId, + }, + }, + { + id: EventId.make(`node:${owner}`), + threadId: owner, + occurredAt: at(1), + type: "node.updated", + payload: { + id: nodeId, + threadId: owner, + runId: ownerRun, + rootNodeId: nodeId, + parentNodeId: null, + kind: "root_turn", + status: "completed", + countsForRun: true, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: at(1), + completedAt: at(2), + }, + }, + { + id: EventId.make(`attempt:${owner}`), + threadId: owner, + occurredAt: at(1), + type: "run-attempt.created", + payload: { + id: attemptId, + runId: ownerRun, + attemptOrdinal: 1, + rootNodeId: nodeId, + providerInstanceId, + providerThreadId: ProviderThreadId.make(`provider-thread:${owner}`), + providerTurnId: null, + reason: "initial", + status: "completed", + startedAt: at(1), + completedAt: at(2), + }, + }, + { + id: EventId.make(`checkpoint:${owner}`), + threadId: owner, + occurredAt: at(2), + type: "checkpoint.captured", + payload: { + id: checkpointId, + threadId: owner, + runId: ownerRun, + nodeId, + scopeId: CheckpointScopeId.make(`scope:${owner}`), + parentCheckpointId: null, + ordinalWithinScope: 1, + appRunOrdinal: 1, + ref: CheckpointRef.make(`refs/t3/${owner}`), + status: "ready", + capturedAt: at(2), + files: [{ path: `${owner}.ts`, kind: "modified", additions: 1, deletions: 0 }], + }, + }, + ]); + } yield* putItems([ - { ...item("inherited", 1, "needle", "assistant_message", parent), runId: parentRun }, + { + ...item("inherited", 1, "needle", "assistant_message", parent), + runId: parentRun, + nodeId: NodeId.make(`node:${parent}`), + }, { ...item("after-fork", 2, "needle", "assistant_message", parent), runId: laterRun }, ]); const child = thread(threadId, projectId); @@ -725,7 +822,10 @@ describe("V2 thread find", () => { }, }, ]); - yield* putItems([item("local", 1, "needle")]); + yield* putItems([ + { ...item("local", 1, "needle"), nodeId: NodeId.make(`node:${threadId}`) }, + item("local-later", 2, "after the history page"), + ]); const inherited = yield* projection.searchThread({ threadId, query: "needle" }); assert.equal(inherited.totalMatches, 2); assert.equal(inherited.match?.entryId, "inherited"); @@ -739,9 +839,12 @@ describe("V2 thread find", () => { encodeThreadHistoryCursor({ snapshotSequence: snapshot.snapshotSequence, sourceThreadId: threadId, - sourceItemId: "local", - position: 2, + sourceItemId: "local-later", + position: 3, }), + undefined, + false, + true, ); assert.deepEqual( history.items, @@ -750,7 +853,34 @@ describe("V2 thread find", () => { .map((row) => ({ ...row, item: projectTurnItemForWire(row.item) })), ); assert.equal(history.items[0]?.sourceThreadId, parent); - assert.equal(history.items.at(-1)?.item.type, "fork"); + assert.equal(history.items[1]?.item.type, "fork"); + assert.equal(history.items.at(-1)?.sourceItemId, "local"); + assert.deepEqual( + { + runs: history.support?.runs.map((run) => [run.id, run.threadId, run.ordinal]), + nodes: history.support?.nodes.map((node) => [node.id, node.threadId]), + attempts: history.support?.attempts.map((attempt) => [attempt.id, attempt.runId]), + checkpoints: history.support?.checkpoints.map((checkpoint) => [ + checkpoint.id, + checkpoint.threadId, + checkpoint.appRunOrdinal, + checkpoint.files, + ]), + }, + { + runs: [[runId, threadId, 1]], + nodes: [[`node:${threadId}`, threadId]], + attempts: [[`attempt:${threadId}`, runId]], + checkpoints: [ + [ + `checkpoint:${threadId}`, + threadId, + 1, + [{ path: `${threadId}.ts`, kind: "modified", additions: 1, deletions: 0 }], + ], + ], + }, + ); yield* putItems( [{ ...item("inherited", 1, "gone", "assistant_message", parent), runId: parentRun }], "parent-update", @@ -759,7 +889,13 @@ describe("V2 thread find", () => { (yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1, ); - }).pipe(Effect.provide(layerTest)), + }).pipe( + Effect.provide( + Layer.mergeAll(EventStore.layer, ProjectStore.layer, projectionLayer).pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), + ), + ), + ), ); it.effect("rejects deleted threads and keeps results scoped to their thread", () => From 50e81b87fa6e3757d3262e96d5d54f3da649d269 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:07:22 +0000 Subject: [PATCH 5/9] fix(server): preserve viewport prompts and thread-local support --- .../src/orchestration-v2/ProjectionStore.ts | 154 ++++++++++-------- .../orchestration-v2/threadHistoryPaging.ts | 30 ++-- 2 files changed, 102 insertions(+), 82 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index f3d7b7038e5d5..5ecb795231a2c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -331,6 +331,7 @@ function historySupportForItems( projection: OrchestrationV2ThreadProjection, items: ReadonlyArray, ) { + items = items.filter((row) => row.sourceThreadId === projection.thread.id); const runIds = new Set(items.map(({ item }) => item.runId)); const runs = projection.runs.filter((run) => runIds.has(run.id)); const attempts = projection.attempts.filter((attempt) => runIds.has(attempt.runId)); @@ -2949,6 +2950,12 @@ export const layer: Layer.Layer = WHEN (SELECT COUNT(*) FROM user_anchors) > 0 THEN -1 ELSE ${window.rowLimit} END + ), prompt_runs AS ( + SELECT run_id FROM selected WHERE run_id IS NOT NULL + UNION + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued','preparing','starting','running','waiting') ), retained AS MATERIALIZED ( SELECT item_rowid, ordinal, turn_item_id FROM selected UNION @@ -2967,43 +2974,36 @@ export const layer: Layer.Layer = AND request.type = 'run_interrupt_request' UNION SELECT prompt.rowid, prompt.ordinal, prompt.turn_item_id - FROM orchestration_v2_projection_turn_items AS prompt - WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 - AND prompt.thread_id = ${threadId} - AND prompt.run_id IN ( - SELECT run_id FROM ( - SELECT run_id FROM selected ORDER BY ordinal ASC, turn_item_id ASC LIMIT 1 - ) - UNION - SELECT run_id FROM orchestration_v2_projection_runs - WHERE thread_id = ${threadId} - AND status IN ('queued','preparing','starting','running','waiting') - ) - AND prompt.rowid = ( + FROM prompt_runs + CROSS JOIN orchestration_v2_projection_turn_items AS prompt ON prompt.rowid = ( SELECT first_input.rowid FROM orchestration_v2_projection_turn_items first_input - WHERE first_input.run_id = prompt.run_id + WHERE first_input.run_id = prompt_runs.run_id + AND first_input.thread_id = ${threadId} AND (first_input.type = 'notification' OR first_input.type = 'user_message' AND json_extract(first_input.payload_json, '$.inputIntent') IN ('turn_start','queued_turn')) ORDER BY first_input.ordinal LIMIT 1 ) + WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 UNION SELECT control.rowid, control.ordinal, control.turn_item_id - FROM orchestration_v2_projection_turn_items AS control + FROM orchestration_v2_projection_runs AS active_run + CROSS JOIN orchestration_v2_projection_turn_items AS control + ON control.run_id = active_run.run_id WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 + AND active_run.thread_id = ${threadId} + AND active_run.status IN ('queued','preparing','starting','running','waiting') AND control.thread_id = ${threadId} - AND ( - (control.status IN ('pending','starting','running','waiting') - AND control.run_id IN ( - SELECT run_id FROM orchestration_v2_projection_runs - WHERE thread_id = ${threadId} - AND status IN ('queued','preparing','starting','running','waiting') - )) - OR (control.type IN ('approval_request','user_input_request') - AND json_extract(control.payload_json, '$.requestId') IN ( - SELECT runtime_request_id FROM orchestration_v2_projection_runtime_requests - WHERE thread_id = ${threadId} AND status IN ('pending','waiting') - )) - ) + AND control.status IN ('pending','starting','running','waiting') + UNION + SELECT control.rowid, control.ordinal, control.turn_item_id + FROM orchestration_v2_projection_runtime_requests AS request + CROSS JOIN orchestration_v2_projection_turn_items AS control + ON control.node_id = request.node_id + WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 + AND request.thread_id = ${threadId} AND request.status IN ('pending','waiting') + AND control.thread_id = ${threadId} + AND control.type IN ('approval_request','user_input_request') + AND json_extract(control.payload_json, '$.requestId') = request.runtime_request_id UNION SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id FROM ( @@ -5167,6 +5167,7 @@ export const layer: Layer.Layer = threadId: ThreadId, items: ReadonlyArray, ) { + items = items.filter((row) => row.sourceThreadId === threadId); const runIds = encodeIdList([ ...new Set(items.flatMap(({ item }) => (item.runId === null ? [] : [item.runId]))), ]); @@ -5176,29 +5177,32 @@ export const layer: Layer.Layer = const checkpointIds = encodeIdList( items.flatMap(({ item }) => ("checkpointId" in item ? [item.checkpointId] : [])), ); - const sourceThreadIds = encodeIdList([...new Set(items.map((row) => row.sourceThreadId))]); const [nodes, checkpoints, runs, attempts] = yield* Effect.all([ sql` WITH RECURSIVE retained(node_id) AS ( SELECT value FROM json_each(${nodeIds}) UNION SELECT json_extract(payload_json, '$.rootNodeId') FROM orchestration_v2_projection_runs - WHERE run_id IN (SELECT value FROM json_each(${runIds})) + WHERE thread_id = ${threadId} AND run_id IN (SELECT value FROM json_each(${runIds})) UNION SELECT root_node_id FROM orchestration_v2_projection_run_attempts - WHERE run_id IN (SELECT value FROM json_each(${runIds})) + WHERE thread_id = ${threadId} AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} AND run_id IN (SELECT value FROM json_each(${runIds})) + ) UNION SELECT node.parent_node_id FROM orchestration_v2_projection_nodes node JOIN retained ON retained.node_id = node.node_id - WHERE node.parent_node_id IS NOT NULL + WHERE node.thread_id = ${threadId} AND node.parent_node_id IS NOT NULL ) SELECT node.payload_json FROM retained CROSS JOIN orchestration_v2_projection_nodes node ON node.node_id = retained.node_id + WHERE node.thread_id = ${threadId} ORDER BY COALESCE(node.started_at, ''), node.node_id `.pipe(Effect.flatMap(decodeRows(decodeNodePayload, threadId))), sql` SELECT payload_json FROM orchestration_v2_projection_checkpoints - WHERE thread_id IN (SELECT value FROM json_each(${sourceThreadIds})) + WHERE thread_id = ${threadId} AND (run_id IN (SELECT value FROM json_each(${runIds})) OR node_id IN (SELECT value FROM json_each(${nodeIds})) OR checkpoint_id IN (SELECT value FROM json_each(${checkpointIds}))) @@ -5206,11 +5210,15 @@ export const layer: Layer.Layer = `.pipe(Effect.flatMap(decodeRows(decodeCheckpointPayload, threadId))), sql` SELECT payload_json FROM orchestration_v2_projection_runs - WHERE run_id IN (SELECT value FROM json_each(${runIds})) ORDER BY ordinal + WHERE thread_id = ${threadId} AND run_id IN (SELECT value FROM json_each(${runIds})) + ORDER BY ordinal `.pipe(Effect.flatMap(decodeRows(decodeRunPayload, threadId))), sql` SELECT payload_json FROM orchestration_v2_projection_run_attempts - WHERE run_id IN (SELECT value FROM json_each(${runIds})) ORDER BY run_id, attempt_ordinal + WHERE thread_id = ${threadId} AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} AND run_id IN (SELECT value FROM json_each(${runIds})) + ) ORDER BY run_id, attempt_ordinal `.pipe(Effect.flatMap(decodeRows(decodeRunAttemptPayload, threadId))), ]); return { nodes, checkpoints, runs, attempts }; @@ -6931,19 +6939,8 @@ export const layerMemory: Layer.Layer = Layer.effect( }), catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), }); - const supports = includeSupport - ? yield* Effect.forEach( - [...new Set(page.items.map((row) => row.sourceThreadId))], - (sourceId) => - service.getThreadSnapshot(sourceId).pipe( - Effect.map(({ projection }) => - historySupportForItems( - projection, - page.items.filter((row) => row.sourceThreadId === sourceId), - ), - ), - ), - ) + const support = includeSupport + ? historySupportForItems(projection, page.items) : undefined; return { ...page, @@ -6951,16 +6948,7 @@ export const layerMemory: Layer.Layer = Layer.effect( items: conversationOnly ? page.items.filter((row) => isConversationHistoryItem(row.item)) : page.items, - ...(supports === undefined - ? {} - : { - support: { - nodes: supports.flatMap((support) => support.nodes), - checkpoints: supports.flatMap((support) => support.checkpoints), - runs: supports.flatMap((support) => support.runs), - attempts: supports.flatMap((support) => support.attempts), - }, - }), + ...(support === undefined ? {} : { support }), }; }), ), @@ -7009,17 +6997,29 @@ export const layerMemory: Layer.Layer = Layer.effect( : anchors.at(-anchorLimit)! : Math.max(0, anchorIndex - options.rowLimit); const windowItems = candidates.slice(start); - const firstRunId = windowItems[0]?.item.runId; - const prompt = - options.deferHistoricalSupport && firstRunId != null - ? candidates.find( - (row, index) => row.item.runId === firstRunId && turnStarts[index], + const windowRunIds = new Set(windowItems.map(({ item }) => item.runId)); + const prompts = options.deferHistoricalSupport + ? candidates.filter((row) => { + if ( + row.item.runId == null || + !windowRunIds.has(row.item.runId) || + !( + row.item.type === "notification" || + (row.item.type === "user_message" && + ["turn_start", "queued_turn"].includes(row.item.inputIntent)) + ) ) - : undefined; + return false; + windowRunIds.delete(row.item.runId); + return !windowItems.includes(row); + }) + : []; const visibleTurnItems = - prompt !== undefined && !windowItems.includes(prompt) - ? [prompt, ...windowItems] - : windowItems; + prompts.length === 0 + ? windowItems + : [...prompts, ...windowItems].sort( + (left, right) => left.position - right.position, + ); const recentRunIds = new Set( snapshot.projection.runs .filter((run) => run.status !== "rolled_back") @@ -7032,6 +7032,19 @@ export const layerMemory: Layer.Layer = Layer.effect( .filter((run) => activeRunStatuses.has(run.status)) .map((run) => run.id), ]); + const fullNodeIds = new Set(visibleTurnItems.map(({ item }) => item.nodeId)); + const fullCheckpointIds = new Set( + visibleTurnItems.flatMap(({ item }) => + "checkpointId" in item ? [item.checkpointId] : [], + ), + ); + const retainCheckpointFiles = ( + checkpoint: OrchestrationV2ThreadProjection["checkpoints"][number], + ) => + ["pending", "capturing"].includes(checkpoint.status) || + (checkpoint.runId !== null && fullRunIds.has(checkpoint.runId)) || + (checkpoint.nodeId !== null && fullNodeIds.has(checkpoint.nodeId)) || + fullCheckpointIds.has(checkpoint.id); return { ...snapshot, projection: { @@ -7042,12 +7055,11 @@ export const layerMemory: Layer.Layer = Layer.effect( checkpoints: snapshot.projection.checkpoints .filter( (checkpoint) => - checkpoint.runId !== null && - (recentRunIds.has(checkpoint.runId) || - fullRunIds.has(checkpoint.runId)), + retainCheckpointFiles(checkpoint) || + (checkpoint.runId !== null && recentRunIds.has(checkpoint.runId)), ) .map((checkpoint) => - fullRunIds.has(checkpoint.runId) + retainCheckpointFiles(checkpoint) ? checkpoint : { ...checkpoint, diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index b2727b5a47dae..62ed4e9c6a955 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -478,6 +478,7 @@ export function buildBoundedThreadProjection(input: { ...(input.viewportWindow ? { checkpoints: input.projection.checkpoints.map((checkpoint) => + ["pending", "capturing"].includes(checkpoint.status) || (checkpoint.runId !== null && retainedRunIds.has(checkpoint.runId)) || (checkpoint.nodeId !== null && retainedNodeIds.has(checkpoint.nodeId)) || retainedCheckpointIds.has(checkpoint.id) || @@ -527,20 +528,27 @@ export function buildBoundedThreadProjection(input: { policy: windowPolicy, rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId), }); - const oldest = window.items[0]; - const prompt = - input.viewportWindow && oldest?.item.runId != null - ? controlProjection.visibleTurnItems.find( - (row) => - row.item.runId === oldest.item.runId && - (isThreadHistoryTurnStart(row.item) || row.item.type === "notification"), - ) - : undefined; + const windowRunIds = new Set(window.items.map(({ item }) => item.runId)); + const visibleItemIds = new Set(window.items.map((row) => row.sourceItemId)); + const prompts = input.viewportWindow + ? controlProjection.visibleTurnItems.filter((row) => { + if ( + row.item.runId == null || + !windowRunIds.has(row.item.runId) || + !(isThreadHistoryTurnStart(row.item) || row.item.type === "notification") + ) { + return false; + } + windowRunIds.delete(row.item.runId); + return !visibleItemIds.has(row.sourceItemId); + }) + : []; // The cursor stays at the row window so paging fills the gap below its prompt. + for (const prompt of prompts) visibleItemIds.add(prompt.sourceItemId); const visibleTurnItems = renumberPositions( - prompt === undefined || window.items.includes(prompt) + prompts.length === 0 ? window.items - : [prompt, ...window.items], + : controlProjection.visibleTurnItems.filter((row) => visibleItemIds.has(row.sourceItemId)), ); const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems); const dependencyTurnItems = retainedInterruptRequestTurnItems( From 60538e379350a0bef152a28cc0b3550d8e161b45 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:09:05 +0000 Subject: [PATCH 6/9] test(server): cover viewport prompt byte budgets --- .../threadHistoryPaging.test.ts | 77 ++++++++++++++++++- 1 file changed, 76 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index c8979fd3bedb9..9574f427fe529 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -188,7 +188,7 @@ describe("threadHistoryPaging", () => { creationSource: "web", inputIntent: "turn_start", messageId: MessageId.make("long-turn-prompt"), - text: "Do the long task", + text: "Do the long task.\n".repeat(375), attachments: [], }, }; @@ -233,6 +233,78 @@ describe("threadHistoryPaging", () => { rows.map((row) => row.sourceItemId), ); expect(older.hasMoreHistory).toBe(false); + + const policy = { maxItems: 75, maxEncodedBytes: 20_000 }; + const encodedBytes = (value: unknown) => Buffer.byteLength(JSON.stringify(value), "utf8"); + expect(encodedBytes(makeProjection([prompt]))).toBeLessThan(policy.maxEncodedBytes); + const smallest = buildBoundedThreadProjection({ + projection, + snapshotSequence: 7, + viewportWindow: true, + policy: { ...policy, maxItems: 1 }, + }); + expect(smallest.projection.visibleTurnItems.map((row) => row.sourceItemId)).toEqual([ + prompt.sourceItemId, + rows.at(-1)!.sourceItemId, + ]); + expect(encodedBytes(smallest.projection)).toBeLessThan(policy.maxEncodedBytes); + expect(smallest.payloadBudgetExceeded).toBe(false); + + const byteBounded = buildBoundedThreadProjection({ + projection, + snapshotSequence: 7, + viewportWindow: true, + policy, + }); + expect.soft(encodedBytes(byteBounded.projection)).toBeLessThanOrEqual(policy.maxEncodedBytes); + expect.soft(byteBounded.payloadBudgetExceeded).toBe(false); + expect(byteBounded.projection.visibleTurnItems[0]?.item).toEqual(prompt.item); + expect(byteBounded.projection.turnItems).toContainEqual(prompt.item); + expect(byteBounded.projection.turnItems).toContainEqual(rows[1]!.item); + const tail = byteBounded.projection.visibleTurnItems.slice(1); + expect(tail.length).toBeGreaterThan(0); + expect(tail.length).toBeLessThan(recent.length); + expect(tail.map((row) => row.sourceItemId)).toEqual( + rows.slice(-tail.length).map((row) => row.sourceItemId), + ); + expect(byteBounded.hasMoreHistory).toBe(true); + expect(decodeThreadHistoryCursor(byteBounded.historyCursor!).si).toBe(tail[0]!.sourceItemId); + let cursor = byteBounded.historyCursor; + const recovered = [...tail]; + const seenCursors = new Set(); + while (cursor !== null) { + expect(seenCursors.has(cursor)).toBe(false); + seenCursors.add(cursor); + const page = selectHistoryPageFromCursor({ + items: rows, + cursor, + snapshotSequence: 7, + policy, + }); + expect(page.items.length).toBeGreaterThan(0); + expect(page.hasMoreHistory).toBe(page.nextCursor !== null); + recovered.unshift(...page.items); + cursor = page.nextCursor; + } + expect(recovered.map((row) => row.sourceItemId)).toEqual(rows.map((row) => row.sourceItemId)); + + const hugePrompt = { ...prompt, item: { ...prompt.item, text: "p".repeat(20_000) } }; + const hugeRows = [hugePrompt, ...rows.slice(1)]; + const oversized = buildBoundedThreadProjection({ + projection: { + ...projection, + visibleTurnItems: hugeRows, + turnItems: hugeRows.map((row) => row.item), + }, + snapshotSequence: 7, + viewportWindow: true, + policy, + }); + expect(encodedBytes(oversized.projection)).toBeGreaterThan(policy.maxEncodedBytes); + expect(oversized.payloadBudgetExceeded).toBe(true); + expect(oversized.projection.visibleTurnItems[0]?.item).toEqual(hugePrompt.item); + expect(oversized.projection.visibleTurnItems.at(-1)?.item).toEqual(rows.at(-1)!.item); + expect(oversized.projection.turnItems).toContainEqual(hugePrompt.item); }); it("keeps background turns with user turns, with main's 150-turn fan-out ceiling", () => { @@ -776,6 +848,9 @@ describe("threadHistoryPaging", () => { expect(bounded.payloadBudgetExceeded).toBe(true); expect(bounded.projection.plans[0]).toEqual(actionable.plans[0]); expect(bounded.projection.visibleTurnItems).toEqual(actionable.visibleTurnItems); + expect(Buffer.byteLength(JSON.stringify(bounded.projection), "utf8")).toBeGreaterThan( + THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes, + ); }); it("keeps agent-only turns visible when checkpoint file lists fill the byte budget", () => { From ae55d13f70d37408d9698cef9ca5a9e0a207bac3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:12:48 +0000 Subject: [PATCH 7/9] fix(server): budget initiating prompts in viewport snapshots --- .../orchestration-v2/threadHistoryPaging.ts | 59 ++++++++++++++----- 1 file changed, 45 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 62ed4e9c6a955..e03b91f56eb0b 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -495,6 +495,21 @@ export function buildBoundedThreadProjection(input: { }; const latestLocalTurnOrdinal = computeLatestLocalTurnOrdinal(input.projection.turnItems); + const promptByRunId = new Map(); + const unbudgetedMessages = new Map( + input.viewportWindow ? controlProjection.messages.map((message) => [message.id, message]) : [], + ); + if (input.viewportWindow) { + for (const row of controlProjection.visibleTurnItems) { + if ( + row.item.runId != null && + !promptByRunId.has(row.item.runId) && + (isThreadHistoryTurnStart(row.item) || row.item.type === "notification") + ) { + promptByRunId.set(row.item.runId, row); + } + } + } const dependencyReserve = (() => { let reserve = 0; for (const item of controlProjection.turnItems) { @@ -502,6 +517,12 @@ export function buildBoundedThreadProjection(input: { reserve += bytesOfJson(item); } } + if (input.viewportWindow) { + for (const message of messagesForBoundedProjection(controlProjection, [])) { + reserve += bytesOfJson(message); + unbudgetedMessages.delete(message.id); + } + } return reserve; })(); const controlBytes = bytesOfJson({ @@ -522,27 +543,37 @@ export function buildBoundedThreadProjection(input: { maxEncodedBytes: windowBudget, }; + const budgetedRowIds = new Set(); + const chargeRow = (row: OrchestrationV2ProjectedTurnItem) => { + if (budgetedRowIds.has(row.sourceItemId)) return 0; + budgetedRowIds.add(row.sourceItemId); + let cost = projectedRowBoundedSnapshotEncodedBytes(row, threadId); + if (isLocalProjectedRow(controlProjection, row) && "messageId" in row.item) { + const message = unbudgetedMessages.get(row.item.messageId); + if (message !== undefined) { + cost += bytesOfJson(message); + unbudgetedMessages.delete(message.id); + } + } + return cost; + }; const window = selectRecentTimelineWindow({ items: controlProjection.visibleTurnItems, snapshotSequence: input.snapshotSequence, policy: windowPolicy, - rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId), + rowEncodedBytes: (row) => { + if (!input.viewportWindow) return projectedRowBoundedSnapshotEncodedBytes(row, threadId); + // Charge a run's prompt with its first admitted row, even when the prompt is outside the tail. + const cost = chargeRow(row); + const prompt = row.item.runId == null ? undefined : promptByRunId.get(row.item.runId); + return cost + (prompt === undefined ? 0 : chargeRow(prompt)); + }, }); const windowRunIds = new Set(window.items.map(({ item }) => item.runId)); const visibleItemIds = new Set(window.items.map((row) => row.sourceItemId)); - const prompts = input.viewportWindow - ? controlProjection.visibleTurnItems.filter((row) => { - if ( - row.item.runId == null || - !windowRunIds.has(row.item.runId) || - !(isThreadHistoryTurnStart(row.item) || row.item.type === "notification") - ) { - return false; - } - windowRunIds.delete(row.item.runId); - return !visibleItemIds.has(row.sourceItemId); - }) - : []; + const prompts = [...promptByRunId.values()].filter( + (row) => windowRunIds.has(row.item.runId) && !visibleItemIds.has(row.sourceItemId), + ); // The cursor stays at the row window so paging fills the gap below its prompt. for (const prompt of prompts) visibleItemIds.add(prompt.sourceItemId); const visibleTurnItems = renumberPositions( From 105726269d1b8de301c1ad55400cc02d8baac3c3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:22:15 +0000 Subject: [PATCH 8/9] test(server): cover checkpoint deferral after viewport byte trimming --- .../threadHistoryPaging.test.ts | 72 +++++++++++++++++-- 1 file changed, 65 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 9574f427fe529..d43f04e45b93b 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -853,7 +853,7 @@ describe("threadHistoryPaging", () => { ); }); - it("keeps agent-only turns visible when checkpoint file lists fill the byte budget", () => { + it("keeps agent-only turns visible and defers checkpoint files outside the final viewport", () => { const items = Array.from({ length: 3 }, (_, turn) => { const first = makeRow(turn * 5); const prompt = { @@ -861,6 +861,7 @@ describe("threadHistoryPaging", () => { item: { ...first.item, type: "user_message" as const, + runId: RunId.make(`run-${turn}`), createdBy: "agent" as const, creationSource: "provider" as const, inputIntent: "turn_start" as const, @@ -869,13 +870,20 @@ describe("threadHistoryPaging", () => { attachments: [], }, } as OrchestrationV2ProjectedTurnItem; - return [prompt, ...Array.from({ length: 4 }, (_, index) => makeRow(turn * 5 + index + 1))]; + return [ + prompt, + ...Array.from({ length: 4 }, (_, index) => { + const row = makeRow(turn * 5 + index + 1); + return { ...row, item: { ...row.item, runId: prompt.item.runId } }; + }), + ]; }).flat(); - const files = Array.from( - { length: 15_000 }, - (_, index) => - `work/ws-${index % 40}/.jj/repo/store/extra/${index.toString(16).padStart(40, "0")}`, - ); + const files = Array.from({ length: 15_000 }, (_, index) => ({ + path: `work/ws-${index % 40}/.jj/repo/store/extra/${index.toString(16).padStart(40, "0")}`, + kind: "modified" as const, + additions: 1, + deletions: 0, + })); const projection = { ...makeProjection(items), checkpoints: [{ id: "checkpoint-1", files }], @@ -886,6 +894,56 @@ describe("threadHistoryPaging", () => { expect(bounded.projection.visibleTurnItems).toHaveLength(items.length); expect(bounded.hasMoreHistory).toBe(false); expect(bounded.payloadBudgetExceeded).toBe(true); + expect(bounded.projection.checkpoints).toEqual(projection.checkpoints); + + const viewportItems = items.map((row) => + row.item.type === "command_execution" + ? { ...row, item: { ...row.item, output: "x".repeat(50_000) } } + : row, + ); + const viewportFiles = files.slice(0, 1_500); + const viewportProjection = { + ...makeProjection(viewportItems), + runs: [0, 1, 2, 3].map((index) => ({ + id: `run-${index}`, + status: index === 3 ? "running" : "completed", + })), + checkpoints: [ + ...[0, 1, 2, 3].map((index) => ({ + id: `checkpoint-${index}`, + runId: `run-${index}`, + nodeId: null, + status: "ready", + files: index === 3 ? files.slice(0, 1) : viewportFiles, + })), + ...["pending", "capturing"].map((status) => ({ + id: `checkpoint-${status}`, + runId: null, + nodeId: null, + status, + files: files.slice(0, 1), + })), + ], + } as unknown as typeof projection; + const viewport = buildBoundedThreadProjection({ + projection: viewportProjection, + snapshotSequence: 1, + viewportWindow: true, + }); + expect(viewport.projection.visibleTurnItems.length).toBeLessThan(viewportItems.length); + expect(new Set(viewport.projection.visibleTurnItems.map((row) => row.item.runId))).toEqual( + new Set([RunId.make("run-1"), RunId.make("run-2")]), + ); + expect.soft(viewport.projection.checkpoints[0]?.files).toHaveLength(0); + expect.soft(viewport.projection.checkpoints[0]?.deferredFileCount).toBe(viewportFiles.length); + expect(viewport.projection.checkpoints.slice(1)).toEqual( + viewportProjection.checkpoints.slice(1), + ); + expect + .soft(Buffer.byteLength(JSON.stringify(viewport.projection), "utf8")) + .toBeLessThanOrEqual(THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes); + expect.soft(viewport.payloadBudgetExceeded).toBe(false); + expect(viewportProjection.checkpoints[0]?.files).toEqual(viewportFiles); }); it("keeps paged historical plan detail in the turn item and only status in its artifact", () => { From 2f304d711bee5391ef6390d7fe5aeee5840b621f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:26:43 +0000 Subject: [PATCH 9/9] fix(server): defer checkpoint files after viewport byte trimming --- .../orchestration-v2/threadHistoryPaging.ts | 64 +++++++++++++------ 1 file changed, 43 insertions(+), 21 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index e03b91f56eb0b..d17c2fc10357d 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -454,6 +454,39 @@ export function boundedTimelineEncodedBytes(input: { return total; } +function checkpointsForViewport( + projection: OrchestrationV2ThreadProjection, + items: ReadonlyArray, +) { + const retainedRunIds = new Set(items.map((item) => item.runId)); + for (const run of projection.runs) { + if (["queued", "preparing", "starting", "running", "waiting"].includes(run.status)) { + retainedRunIds.add(run.id); + } + } + const retainedNodeIds = new Set(items.map((item) => item.nodeId)); + for (const node of projection.nodes) { + if (["pending", "running", "waiting"].includes(node.status)) { + retainedNodeIds.add(node.id); + } + } + for (const request of projection.runtimeRequests) { + if (request.status === "pending") retainedNodeIds.add(request.nodeId); + } + const retainedCheckpointIds = new Set( + items.flatMap((item) => ("checkpointId" in item ? [item.checkpointId] : [])), + ); + return projection.checkpoints.map((checkpoint) => + ["pending", "capturing"].includes(checkpoint.status) || + (checkpoint.runId !== null && retainedRunIds.has(checkpoint.runId)) || + (checkpoint.nodeId !== null && retainedNodeIds.has(checkpoint.nodeId)) || + retainedCheckpointIds.has(checkpoint.id) || + checkpoint.deferredFileCount !== undefined + ? checkpoint + : { ...checkpoint, files: [], deferredFileCount: checkpoint.files.length }, + ); +} + export function buildBoundedThreadProjection(input: { readonly projection: OrchestrationV2ThreadProjection; readonly snapshotSequence: number; @@ -462,29 +495,13 @@ export function buildBoundedThreadProjection(input: { }): BoundedProjectionResult { const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY; const threadId = input.projection.thread.id; - const recentRows = input.projection.visibleTurnItems.slice(-policy.maxItems); - const retainedRunIds = new Set(recentRows.map(({ item }) => item.runId)); - for (const run of input.projection.runs) { - if (["queued", "preparing", "starting", "running", "waiting"].includes(run.status)) { - retainedRunIds.add(run.id); - } - } - const retainedNodeIds = new Set(recentRows.map(({ item }) => item.nodeId)); - const retainedCheckpointIds = new Set( - recentRows.flatMap(({ item }) => ("checkpointId" in item ? [item.checkpointId] : [])), - ); const controlProjection = { ...input.projection, ...(input.viewportWindow ? { - checkpoints: input.projection.checkpoints.map((checkpoint) => - ["pending", "capturing"].includes(checkpoint.status) || - (checkpoint.runId !== null && retainedRunIds.has(checkpoint.runId)) || - (checkpoint.nodeId !== null && retainedNodeIds.has(checkpoint.nodeId)) || - retainedCheckpointIds.has(checkpoint.id) || - checkpoint.deferredFileCount !== undefined - ? checkpoint - : { ...checkpoint, files: [], deferredFileCount: checkpoint.files.length }, + checkpoints: checkpointsForViewport( + input.projection, + input.projection.visibleTurnItems.slice(-policy.maxItems).map(({ item }) => item), ), } : {}), @@ -593,6 +610,7 @@ export function buildBoundedThreadProjection(input: { for (const item of dependencyTurnItems) { turnItemById.set(String(item.id), item); } + const checkpointItems = visibleTurnItems.map(({ item }) => item); if (input.viewportWindow) { const activeRunIds = new Set( controlProjection.runs @@ -616,6 +634,7 @@ export function buildBoundedThreadProjection(input: { ("requestId" in item && pendingRequestIds.has(item.requestId)) ) { turnItemById.set(String(item.id), item); + checkpointItems.push(item); } } } @@ -653,6 +672,10 @@ export function buildBoundedThreadProjection(input: { const projection = { ...controlProjection, + // Byte trimming can remove runs that were eligible when reserving control state. + ...(input.viewportWindow + ? { checkpoints: checkpointsForViewport(input.projection, checkpointItems) } + : {}), plans, contextHandoffs, messages: messagesForBoundedProjection(controlProjection, turnItems), @@ -664,7 +687,6 @@ export function buildBoundedThreadProjection(input: { historyCursor: window.nextCursor, hasMoreHistory: window.hasMoreHistory, latestLocalTurnOrdinal, - payloadBudgetExceeded: - controlBytes > policy.maxEncodedBytes || bytesOfJson(projection) > policy.maxEncodedBytes, + payloadBudgetExceeded: bytesOfJson(projection) > policy.maxEncodedBytes, }; }