diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 74377160a8204..8c1a6121fc953 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 OrchestrationV2ThreadHistoryPage, type OrchestrationV2ThreadShellSnapshot, NodeId, ProjectId, @@ -1091,7 +1092,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); - it.effect("pages complete user turns through SQL regardless of tool count or payload size", () => + it.effect("pages oversized user turns individually through SQL without splitting them", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const sql = yield* SqlClient.SqlClient; @@ -1236,11 +1237,11 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { projection: initial.projection, snapshotSequence: 0, }); - assert.lengthOf(bounded.projection.visibleTurnItems, 10 * 102); - assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, allIds[35 * 102]); + assert.lengthOf(bounded.projection.visibleTurnItems, 102); + assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, allIds[44 * 102]); const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId)); let cursor = bounded.historyCursor; - for (const turns of [20, 15]) { + for (let turn = 44; turn >= 1; turn -= 1) { assert.isNotNull(cursor); const anchor = decodeThreadHistoryCursor(cursor!); const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { @@ -1254,7 +1255,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { cursor: cursor!, snapshotSequence: 0, }); - assert.lengthOf(page.items, turns * 102); + assert.lengthOf(page.items, 102); + assert.strictEqual(page.items[0]?.sourceItemId, allIds[(turn - 1) * 102]); loaded.unshift(...page.items.map((row) => String(row.sourceItemId))); cursor = page.nextCursor; } @@ -1423,11 +1425,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { projection: initial.projection, snapshotSequence: 0, }); - assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, startIds[262]); + assert.lengthOf(bounded.projection.visibleTurnItems, 75); + assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, startIds[393]); const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId)); let cursor = bounded.historyCursor; for (let pages = 0; cursor !== null; pages += 1) { - assert.isBelow(pages, 5); + assert.isBelow(pages, 25); const anchor = decodeThreadHistoryCursor(cursor); const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 77, @@ -1441,6 +1444,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { cursor, snapshotSequence: 0, }); + assert.isAtMost(page.items.length, THREAD_HISTORY_PAGE_POLICY.maxItems); loaded.unshift(...page.items.map((row) => String(row.sourceItemId))); cursor = page.nextCursor; } @@ -2119,7 +2123,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { snapshotSequence: cancelledSuffixSnapshot.snapshotSequence, }); assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202); - assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75); + assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 1); assert.isFalse( cancelledSuffixPage.projection.visibleTurnItems.some((row) => String(row.sourceItemId).includes(":cancelled:"), @@ -2130,6 +2134,28 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { (row) => row.sourceItemId === runlessQueuedId, ), ); + assert.isTrue(cancelledSuffixPage.hasMoreHistory); + assert.isNotNull(cancelledSuffixPage.historyCursor); + const queuedSuffixIds = cancelledSuffixPage.projection.visibleTurnItems.map((row) => + String(row.sourceItemId), + ); + let queuedCursor: string | null = cancelledSuffixPage.historyCursor; + for (let pages = 0; queuedCursor !== null; pages += 1) { + assert.isBelow(pages, 14); + const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore.getThreadHistoryPage( + threadId, + queuedCursor, + ); + queuedSuffixIds.unshift(...page.items.map((row) => String(row.sourceItemId))); + queuedCursor = page.nextCursor; + } + assert.deepEqual(queuedSuffixIds, [ + ...Array.from( + { length: 1_000 }, + (_, index) => `turn-item:bounded-sql-history:${index + 1}`, + ), + runlessQueuedId, + ]); const interruptRunId = RunId.make("run:bounded-sql-history:interrupt"); const interruptNodeId = NodeId.make("node:bounded-sql-history:interrupt"); @@ -5566,6 +5592,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { projection: yield* projectionStore .getThreadSnapshotWindow(nestedThreadId, { rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2, + userTurnLimit: 10, }) .pipe(Effect.map((snapshot) => snapshot.projection)), snapshotSequence: 1, @@ -5582,6 +5609,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const decoded = decodeThreadHistoryCursor(nestedCursor); const snapshot = yield* projectionStore.getThreadSnapshotWindow(nestedThreadId, { rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2, + userTurnLimit: 20, anchorItemId: TurnItemId.make(decoded.si), anchorThreadId: ThreadId.make(decoded.st), }); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa950..375755d4f3cb6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -88,9 +88,7 @@ import { threadHistoryTurnStarts, THREAD_HISTORY_MAX_RAW_TURNS, selectHistoryPageFromCursor, - selectRecentTimelineWindow, - THREAD_HISTORY_PAGE_POLICY, - OLDER_THREAD_USER_TURN_LIMIT, + projectedRowEncodedBytes, } from "./threadHistoryPaging.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( @@ -5073,14 +5071,15 @@ export const layer: Layer.Layer = FROM orchestration_events WHERE application_event_version = 2 AND aggregate_kind = 'thread' AND stream_id = ${threadId}`; const snapshotSequence = revisions[0]?.sequence ?? 0; + const indexedItems = index.visible.map((row, position) => ({ + ...row, + position, + canonicalPosition: position, + })); const page = yield* Effect.try({ try: () => selectHistoryPageFromCursor({ - items: index.visible.map((row, position) => ({ - ...row, - position, - canonicalPosition: position, - })), + items: indexedItems, cursor, snapshotSequence, rowEncodedBytes: () => 0, @@ -5088,16 +5087,8 @@ export const layer: Layer.Layer = }), catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), }); - const end = (page.items.at(-1)?.canonicalPosition ?? -1) + 1; - const hasUserTurns = index.visible - .slice(0, end) - .some((row) => isThreadHistoryUserTurn(row.item)); - const materialized = - conversationOnly && hasUserTurns - ? page.items.filter((row) => isConversationHistoryItem(row.item)) - : page.items; const sources = yield* encodeTimelineSources( - materialized + page.items .filter((row) => !row.synthetic) .map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })), ); @@ -5113,7 +5104,7 @@ export const layer: Layer.Layer = const byIdentity = new Map( payloads.map((row) => [`${row.thread_id}:${row.turn_item_id}`, row.payload_json]), ); - const items = yield* Effect.forEach(materialized, (row) => + const items = yield* Effect.forEach(page.items, (row) => Effect.gen(function* () { const payload = byIdentity.get(`${row.sourceThreadId}:${row.sourceItemId}`); const item = @@ -5130,29 +5121,32 @@ export const layer: Layer.Layer = }; }), ); - if (hasUserTurns) - return { - ...page, - snapshotSequence, - items: items.map((row, position) => ({ ...row, position })), - }; - const bounded = selectRecentTimelineWindow({ - items, + const wireRowsByPosition = new Map( + items.map((row) => [ + row.position, + { row, encodedBytes: projectedRowEncodedBytes(row) }, + ]), + ); + const bounded = selectHistoryPageFromCursor({ + items: indexedItems, + cursor, + throughEntryId, snapshotSequence, - policy: { - ...THREAD_HISTORY_PAGE_POLICY, - maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, - maxItems: - THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2), - }, + rowEncodedBytes: (row) => + wireRowsByPosition.get(row.canonicalPosition)?.encodedBytes ?? 0, }); + // Non-negative wire costs can only trim the suffix materialized by the zero-cost pass. + const selected = bounded.items.map((row) => ({ + ...wireRowsByPosition.get(row.canonicalPosition)!.row, + position: row.position, + })); 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, + ? selected.filter((row) => isConversationHistoryItem(row.item)) + : selected, + nextCursor: bounded.nextCursor, + hasMoreHistory: bounded.hasMoreHistory, }; }), ) diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index 0121227f6e86a..c99a2a39b2f6f 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -986,6 +986,84 @@ describe("V2 history materialization", () => { }).pipe(Effect.provide(layerTest)), ); + it.effect.each(["wake", "steering"] as const)( + "preserves full-history turn boundaries when paging %s notifications in SQL", + (kind) => + Effect.gen(function* () { + const projection = yield* setup; + const notificationRun = kind === "wake" ? RunId.make("run:wake") : runId; + if (kind === "wake") yield* commit([run(threadId, notificationRun, 2)]); + const notification = { + ...item("notification", kind === "wake" ? 80 : 82, ""), + runId: notificationRun, + type: "notification" as const, + source: { kind: "monitor" as const }, + outcome: "completed" as const, + summary: kind, + }; + const newestTurn = + kind === "wake" + ? [ + notification, + ...Array.from({ length: 200 }, (_, index) => ({ + ...item(`wake-tool:${index}`, 81 + index, ""), + runId: notificationRun, + type: "command_execution" as const, + input: "pwd", + output: "/work", + exitCode: 0, + })), + ] + : [ + { + ...item("queued", 80, "next turn", "user_message"), + runId: null, + inputIntent: "queued_turn" as const, + }, + item("answer:before-steer", 81, "a".repeat(700_000)), + notification, + item("answer:after-steer", 83, "b".repeat(700_000)), + ]; + yield* putItems([ + item("user", 0, "initial turn", "user_message"), + ...Array.from({ length: 79 }, (_, index) => item(`older:${index}`, index + 1, "work")), + ...newestTurn, + ]); + const snapshot = yield* projection.getThreadSnapshot(threadId); + const rows = snapshot.projection.visibleTurnItems.map((row) => ({ + ...row, + item: projectTurnItemForWire(row.item), + })); + const cursor = encodeThreadHistoryCursor({ + snapshotSequence: snapshot.snapshotSequence, + sourceThreadId: threadId, + sourceItemId: "absent", + position: rows.length, + }); + const expected = selectHistoryPageFromCursor({ + items: rows, + cursor, + snapshotSequence: snapshot.snapshotSequence, + }); + const expectedCount = kind === "wake" ? 201 : 4; + const expectedFirstId = kind === "wake" ? "notification" : "queued"; + assert.equal(expected.items.length, expectedCount); + assert.equal(expected.items[0]?.sourceItemId, expectedFirstId); + assert.deepEqual( + expected.items.map((row) => row.sourceItemId), + newestTurn.map((entry) => entry.id), + ); + assert.isTrue(expected.hasMoreHistory); + + const page = yield* projection.getThreadHistoryPage(threadId, cursor); + assert.deepEqual( + { count: page.items.length, firstId: page.items[0]?.sourceItemId }, + { count: expectedCount, firstId: expectedFirstId }, + ); + assert.deepEqual(page, { snapshotSequence: snapshot.snapshotSequence, ...expected }); + }).pipe(Effect.provide(layerTest)), + ); + it.effect("enforces fallback item and byte limits and always admits an oversized row", () => Effect.gen(function* () { const projection = yield* setup; diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 0b58163bbdff1..9dc85401aa938 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -67,6 +67,27 @@ function makeRow( }; } +function makeTurnStartRow( + index: number, + createdBy: "user" | "agent" = "user", +): OrchestrationV2ProjectedTurnItem { + const row = makeRow(index); + if (row.item.type !== "command_execution") throw new Error("Expected command fixture"); + return { + ...row, + item: { + ...row.item, + type: "user_message", + createdBy, + creationSource: "provider", + inputIntent: "turn_start", + messageId: MessageId.make(`prompt-${index}`), + text: `Prompt ${index}`, + attachments: [], + }, + }; +} + function interruptItem( id: string, type: "run_interrupt_request" | "run_interrupt_result", @@ -177,30 +198,18 @@ describe("threadHistoryPaging", () => { 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); - if (row.item.type !== "command_execution") throw new Error("Expected command fixture"); - const prompt: OrchestrationV2ProjectedTurnItem = { - ...row, - item: { - ...row.item, - type: "user_message", - createdBy: turn === 0 ? "user" : "agent", - creationSource: "provider", - inputIntent: "turn_start", - messageId: MessageId.make(`prompt-${turn}`), - text: `Prompt ${turn}`, - attachments: [], - }, - }; + const prompt = makeTurnStartRow(turn * 2, turn === 0 ? "user" : "agent"); return [prompt, makeRow(turn * 2 + 1)]; }).flat(); - const first = selectRecentTimelineWindow({ items, snapshotSequence: 1 }); + const policy = { maxUserTurns: 10, maxItems: 1_000, maxEncodedBytes: 10_000_000 }; + const first = selectRecentTimelineWindow({ items, snapshotSequence: 1, policy }); expect(first.items).toHaveLength(300); expect(first.items[0]?.sourceItemId).toBe("item-22"); const older = selectHistoryPageFromCursor({ items, cursor: first.nextCursor!, snapshotSequence: 1, + policy, }); expect(older.items).toHaveLength(22); expect(older.hasMoreHistory).toBe(false); @@ -243,13 +252,15 @@ describe("threadHistoryPaging", () => { inRun(run * 4 + 2, run, notification(`Steer ${run}`)), inRun(run * 4 + 3, run), ]).flat(); - const first = selectRecentTimelineWindow({ items, snapshotSequence: 1 }); + const policy = { maxUserTurns: 10, maxItems: 1_000, maxEncodedBytes: 10_000_000 }; + const first = selectRecentTimelineWindow({ items, snapshotSequence: 1, policy }); expect(first.items).toHaveLength(150 * 4 - 1); expect(first.items[0]?.sourceItemId).toBe(`item-${11 * 4 + 1}`); const older = selectHistoryPageFromCursor({ items, cursor: first.nextCursor!, snapshotSequence: 1, + policy, }); expect([...older.items, ...first.items].map((row) => row.sourceItemId)).toEqual( items.map((row) => row.sourceItemId), @@ -355,6 +366,80 @@ describe("threadHistoryPaging", () => { expect(older.nextCursor).toBeNull(); }); + it.each(["item", "byte", "user-turn"] as const)( + "applies the %s budget to complete anchored turns across history pages", + (budget) => { + const items = Array.from({ length: 5 }, (_, turn) => [ + makeTurnStartRow(turn * 2), + makeRow(turn * 2 + 1, { outputBytes: 2_000 }), + ]).flat(); + const policy = { + maxUserTurns: budget === "user-turn" ? 2 : 10, + maxItems: budget === "item" ? 5 : 100, + maxEncodedBytes: + budget === "byte" + ? items.slice(-4).reduce((total, row) => total + projectedRowEncodedBytes(row), 0) + : 1_000_000, + }; + const recent = selectRecentTimelineWindow({ items, snapshotSequence: 1, policy }); + expect(recent.items.map((row) => row.sourceItemId)).toEqual([ + "item-6", + "item-7", + "item-8", + "item-9", + ]); + expect(recent.hasMoreHistory).toBe(true); + + const older = selectHistoryPageFromCursor({ + items, + cursor: recent.nextCursor!, + snapshotSequence: 1, + policy, + }); + expect(older.items.map((row) => row.sourceItemId)).toEqual([ + "item-2", + "item-3", + "item-4", + "item-5", + ]); + expect(older.hasMoreHistory).toBe(true); + + const oldest = selectHistoryPageFromCursor({ + items, + cursor: older.nextCursor!, + snapshotSequence: 1, + policy, + }); + expect(oldest.items.map((row) => row.sourceItemId)).toEqual(["item-0", "item-1"]); + expect(oldest.hasMoreHistory).toBe(false); + expect(oldest.nextCursor).toBeNull(); + }, + ); + + it("keeps a complete 150,000-row turn over both budgets and can page past it", () => { + const items = Array.from({ length: 150_002 }, (_, index) => + index === 0 || index === 2 ? makeTurnStartRow(index) : makeRow(index), + ); + const policy = { maxUserTurns: 10, maxItems: 1, maxEncodedBytes: 100 }; + const page = selectRecentTimelineWindow({ items, snapshotSequence: 1, policy }); + + expect(page.items).toHaveLength(150_000); + expect( + page.items.every((row, index) => row.sourceItemId === items[index + 2]?.sourceItemId), + ).toBe(true); + expect(page.hasMoreHistory).toBe(true); + + const older = selectHistoryPageFromCursor({ + items, + cursor: page.nextCursor!, + snapshotSequence: 1, + policy, + }); + expect(older.items.map((row) => row.sourceItemId)).toEqual(["item-0", "item-1"]); + expect(older.hasMoreHistory).toBe(false); + expect(older.nextCursor).toBeNull(); + }); + it("recovers identity-miss cursors when position is past a shrunken timeline", () => { const items = Array.from({ length: 5 }, (_, index) => makeRow(index)); const cursor = encodeThreadHistoryCursor({ @@ -565,6 +650,30 @@ describe("threadHistoryPaging", () => { ); }); + it("bounds anchored snapshots using the duplicated local-row cost", () => { + const rows = Array.from({ length: 4 }, (_, turn) => [ + makeTurnStartRow(turn * 2), + makeRow(turn * 2 + 1, { outputBytes: 2_000 }), + ]).flat(); + const maxEncodedBytes = 12_000; + const bounded = buildBoundedThreadProjection({ + projection: makeProjection(rows), + snapshotSequence: 1, + policy: { maxUserTurns: 10, maxItems: 50, maxEncodedBytes }, + }); + + expect(bounded.projection.visibleTurnItems.length).toBeLessThan(rows.length); + expect(bounded.projection.visibleTurnItems[0]?.item.type).toBe("user_message"); + expect( + boundedTimelineEncodedBytes({ + visibleTurnItems: bounded.projection.visibleTurnItems, + turnItems: bounded.projection.turnItems, + }), + ).toBeLessThanOrEqual(maxEncodedBytes); + expect(bounded.payloadBudgetExceeded).toBe(false); + expect(bounded.hasMoreHistory).toBe(true); + }); + it("allows a single pathological bounded row to exceed the configured cap", () => { const huge = makeRow(0, { outputBytes: 50_000 }); const full = makeProjection([huge]); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index ee675856477cc..8a22dca0b9291 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -6,10 +6,7 @@ import type { TurnItemId, } from "@t3tools/contracts"; -/** - * Item/byte budgets only apply to histories without turn starts; tool activity - * must not split an ordinary conversation turn. - */ +/** Apply page budgets without splitting turns in histories with a user-created anchor. */ export const THREAD_HISTORY_PAGE_POLICY = { maxUserTurns: 10, maxItems: 75, @@ -189,8 +186,8 @@ export function isThreadHistoryUserTurn(item: HistoryRow["item"]): boolean { } /** - * Histories without turn starts use row budgets and always admit at least one - * row so oversized items cannot deadlock paging. + * Admit complete turns when a user-created anchor exists, otherwise admit rows. + * Always include the newest turn or row so oversized content cannot stall paging. */ function selectOlderTimelinePage(input: { readonly items: ReadonlyArray; @@ -206,41 +203,39 @@ function selectOlderTimelinePage(input: { return { items: [], nextCursor: null, hasMoreHistory: false }; } - const selected: Row[] = []; + let selectedStart = end; let encodedBytes = 0; + let turnEncodedBytes = 0; let userTurns = 0; let rawTurns = 0; const candidates = input.items.slice(0, end); const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item)); - const turnLimit = candidates.some((row) => isThreadHistoryUserTurn(row.item)) - ? policy.maxUserTurns - : undefined; + const hasUserTurns = candidates.some((row) => isThreadHistoryUserTurn(row.item)); for (let index = end - 1; index >= 0; index -= 1) { const row = input.items[index]!; - const rowBytes = turnLimit === undefined ? rowCost(row) : 0; + turnEncodedBytes += rowCost(row); + if (hasUserTurns && index > 0 && !turnStarts[index]) continue; + if ( - selected.length > 0 && - (turnLimit === undefined - ? selected.length >= policy.maxItems || encodedBytes + rowBytes > policy.maxEncodedBytes - : userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS) + selectedStart < end && + (end - index > policy.maxItems || + encodedBytes + turnEncodedBytes > policy.maxEncodedBytes || + (hasUserTurns && + ((policy.maxUserTurns !== undefined && userTurns >= policy.maxUserTurns) || + rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS))) ) { break; } - selected.push(row); - encodedBytes += rowBytes; + selectedStart = index; + encodedBytes += turnEncodedBytes; + turnEncodedBytes = 0; if (isThreadHistoryUserTurn(row.item)) userTurns += 1; if (turnStarts[index]) rawTurns += 1; } - selected.reverse(); + const selected = input.items.slice(selectedStart, end); const oldest = selected[0]; - const oldestIndex = oldest - ? input.items.findIndex( - (row) => - row.sourceThreadId === oldest.sourceThreadId && row.sourceItemId === oldest.sourceItemId, - ) - : -1; - const hasMoreHistory = oldestIndex > 0; + const hasMoreHistory = selectedStart > 0; const nextCursor = hasMoreHistory && oldest ? encodeThreadHistoryCursor({ diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index ade826e53f38b..adc3bc1fc5b4c 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1899,6 +1899,7 @@ const layerWsRpc = ( threadManagement .getThreadSnapshotWindow(input.threadId, { rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, }) .pipe( Effect.map((snapshot) =>