From 875b02835e7f9ec5293179e9259593927607e4a6 Mon Sep 17 00:00:00 2001 From: connorwalker Date: Sun, 4 Oct 2026 22:30:03 +0100 Subject: [PATCH 1/7] fix(server): enforce history page limits at turn boundaries Timelines with turn anchors skipped the 75-item and 1 MiB page budgets, so a long thread could send a very large history page to mobile clients. Apply every limit at turn boundaries instead, keeping turns intact and always admitting the newest turn so an oversized turn cannot stall paging. Co-Authored-By: Claude Sonnet 5.5 --- .../threadHistoryPaging.test.ts | 103 +++++++++++++++--- .../orchestration-v2/threadHistoryPaging.ts | 76 ++++++++----- 2 files changed, 138 insertions(+), 41 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 9c1115ca0df7f..e9b2fc8aa78b7 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); @@ -308,6 +317,44 @@ describe("threadHistoryPaging", () => { expect(older.nextCursor).toBeNull(); }); + it("applies item and byte budgets to anchored histories at turn boundaries", () => { + const items = [ + makeTurnStartRow(0), + makeRow(1), + makeTurnStartRow(2), + makeRow(3), + makeTurnStartRow(4), + makeRow(5), + ]; + const page = selectRecentTimelineWindow({ + items, + snapshotSequence: 1, + policy: { maxUserTurns: 10, maxItems: 3, maxEncodedBytes: 1_000 }, + rowEncodedBytes: () => 400, + }); + + expect(page.items.map((row) => String(row.sourceItemId))).toEqual(["item-4", "item-5"]); + expect(page.items[0]?.item.type).toBe("user_message"); + expect(page.hasMoreHistory).toBe(true); + }); + + it("keeps one complete anchored turn when that turn exceeds the page budget", () => { + const items = [ + makeTurnStartRow(0), + makeRow(1), + makeTurnStartRow(2), + makeRow(3, { outputBytes: 50_000 }), + ]; + const page = selectRecentTimelineWindow({ + items, + snapshotSequence: 1, + policy: { maxUserTurns: 10, maxItems: 1, maxEncodedBytes: 100 }, + }); + + expect(page.items.map((row) => String(row.sourceItemId))).toEqual(["item-2", "item-3"]); + expect(page.hasMoreHistory).toBe(true); + }); + it("recovers identity-miss cursors when position is past a shrunken timeline", () => { const items = Array.from({ length: 5 }, (_, index) => makeRow(index)); const cursor = encodeThreadHistoryCursor({ @@ -533,6 +580,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 a2e6c87e20df7..0a3f2324f8eff 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"; -/** - * Match the V1 conversation windows. Item/byte budgets only apply to histories - * without turn starts; tool activity must not split an ordinary conversation turn. - */ +/** Match the V1 conversation windows without splitting a conversation turn. */ export const THREAD_HISTORY_PAGE_POLICY = { maxUserTurns: 10, maxItems: 75, @@ -163,9 +160,10 @@ export function isThreadHistoryUserTurn(item: OrchestrationV2TurnItem): boolean } /** - * Walk a chronological timeline backward from the exclusive end, collecting rows - * through complete user turns. Histories without turn starts use row budgets - * and always admit at least one row so oversized items cannot deadlock paging. + * Walk a chronological timeline backward from the exclusive end. Anchored + * histories apply every limit at turn boundaries so pages keep turns intact. + * Every history admits its newest row or turn so oversized content cannot + * deadlock paging. */ function selectOlderTimelinePage(input: { readonly items: ReadonlyArray; @@ -185,26 +183,54 @@ function selectOlderTimelinePage(input: { let encodedBytes = 0; let userTurns = 0; let rawTurns = 0; - const turnLimit = input.items.slice(0, end).some((row) => isThreadHistoryUserTurn(row.item)) - ? policy.maxUserTurns - : undefined; - for (let index = end - 1; index >= 0; index -= 1) { - const row = input.items[index]!; - const rowBytes = turnLimit === undefined ? rowCost(row) : 0; - if ( - selected.length > 0 && - (turnLimit === undefined - ? selected.length >= policy.maxItems || encodedBytes + rowBytes > policy.maxEncodedBytes - : userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS) - ) { - break; + const hasUserTurns = input.items.slice(0, end).some((row) => isThreadHistoryUserTurn(row.item)); + const turnLimit = policy.maxUserTurns; + + if (hasUserTurns) { + let turnEnd = end; + for (let index = end - 1; index >= -1; index -= 1) { + const startsTurn = index >= 0 && isThreadHistoryTurnStart(input.items[index]!.item); + if (index >= 0 && !startsTurn) continue; + + const turnStart = index < 0 ? 0 : index; + if (turnStart === turnEnd) continue; + const turnRows = input.items.slice(turnStart, turnEnd); + const turnBytes = turnRows.reduce((total, row) => total + rowCost(row), 0); + const turnUserCount = startsTurn && isThreadHistoryUserTurn(input.items[index]!.item) ? 1 : 0; + const turnRawCount = startsTurn ? 1 : 0; + const hasSelectedTurn = selected.length > 0; + if ( + hasSelectedTurn && + (selected.length + turnRows.length > policy.maxItems || + encodedBytes + turnBytes > policy.maxEncodedBytes || + (turnLimit !== undefined && userTurns >= turnLimit) || + rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS) + ) { + break; + } + + selected.unshift(...turnRows); + encodedBytes += turnBytes; + userTurns += turnUserCount; + rawTurns += turnRawCount; + turnEnd = turnStart; + if (turnStart === 0) break; + } + } else { + for (let index = end - 1; index >= 0; index -= 1) { + const row = input.items[index]!; + const rowBytes = rowCost(row); + if ( + selected.length > 0 && + (selected.length >= policy.maxItems || encodedBytes + rowBytes > policy.maxEncodedBytes) + ) { + break; + } + selected.push(row); + encodedBytes += rowBytes; } - selected.push(row); - encodedBytes += rowBytes; - if (isThreadHistoryUserTurn(row.item)) userTurns += 1; - if (isThreadHistoryTurnStart(row.item)) rawTurns += 1; + selected.reverse(); } - selected.reverse(); const oldest = selected[0]; const oldestIndex = oldest From 6b0e4f0bc60cef20c271b54fb8f7f6d77442d363 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 15:52:31 +0000 Subject: [PATCH 2/7] test(server): distinguish history turn paging budgets --- .../threadHistoryPaging.test.ts | 100 ++++++++++++------ 1 file changed, 69 insertions(+), 31 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 9053b5b325791..9dc85401aa938 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -252,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), @@ -364,42 +366,78 @@ describe("threadHistoryPaging", () => { expect(older.nextCursor).toBeNull(); }); - it("applies item and byte budgets to anchored histories at turn boundaries", () => { - const items = [ - makeTurnStartRow(0), - makeRow(1), - makeTurnStartRow(2), - makeRow(3), - makeTurnStartRow(4), - makeRow(5), - ]; - const page = selectRecentTimelineWindow({ - items, - snapshotSequence: 1, - policy: { maxUserTurns: 10, maxItems: 3, maxEncodedBytes: 1_000 }, - rowEncodedBytes: () => 400, - }); + 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.map((row) => String(row.sourceItemId))).toEqual(["item-4", "item-5"]); - expect(page.items[0]?.item.type).toBe("user_message"); + 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); - }); - it("keeps one complete anchored turn when that turn exceeds the page budget", () => { - const items = [ - makeTurnStartRow(0), - makeRow(1), - makeTurnStartRow(2), - makeRow(3, { outputBytes: 50_000 }), - ]; - const page = selectRecentTimelineWindow({ + const older = selectHistoryPageFromCursor({ items, + cursor: page.nextCursor!, snapshotSequence: 1, - policy: { maxUserTurns: 10, maxItems: 1, maxEncodedBytes: 100 }, + policy, }); - - expect(page.items.map((row) => String(row.sourceItemId))).toEqual(["item-2", "item-3"]); - expect(page.hasMoreHistory).toBe(true); + 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", () => { From e44831b043d8eca49ac3c0be8ecc9176a3c4cb50 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 15:56:15 +0000 Subject: [PATCH 3/7] fix(server): bound history pages at complete turns --- .../orchestration-v2/threadHistoryPaging.ts | 45 +++++++++---------- 1 file changed, 20 insertions(+), 25 deletions(-) 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({ From 5f8909fa8e5f88574480953bdebe82d36ff4fa93 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 16:11:06 +0000 Subject: [PATCH 4/7] test(server): align SQL history fixtures with complete-turn budgets --- .../orchestration-v2/ProjectionStore.test.ts | 40 +++++++++++++++---- 1 file changed, 32 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 74377160a8204..0e25d79ff78f3 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1091,7 +1091,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 +1236,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 +1254,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 +1424,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 +1443,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 +2122,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 +2133,25 @@ 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 = cancelledSuffixPage.historyCursor; + for (let pages = 0; queuedCursor !== null; pages += 1) { + assert.isBelow(pages, 14); + const page = 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 +5588,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 +5605,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), }); From 6f7c1273987631ef925b542bf7344c2b37c908b8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 16:16:01 +0000 Subject: [PATCH 5/7] fix(server): enforce byte budgets on SQL history pages --- .../orchestration-v2/ProjectionStore.test.ts | 8 +++++-- .../src/orchestration-v2/ProjectionStore.ts | 22 ++++--------------- apps/server/src/ws.ts | 1 + 3 files changed, 11 insertions(+), 20 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 0e25d79ff78f3..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, @@ -2138,10 +2139,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const queuedSuffixIds = cancelledSuffixPage.projection.visibleTurnItems.map((row) => String(row.sourceItemId), ); - let queuedCursor = cancelledSuffixPage.historyCursor; + let queuedCursor: string | null = cancelledSuffixPage.historyCursor; for (let pages = 0; queuedCursor !== null; pages += 1) { assert.isBelow(pages, 14); - const page = yield* projectionStore.getThreadHistoryPage(threadId, queuedCursor); + const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore.getThreadHistoryPage( + threadId, + queuedCursor, + ); queuedSuffixIds.unshift(...page.items.map((row) => String(row.sourceItemId))); queuedCursor = page.nextCursor; } diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa950..098e10b7fca03 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -90,7 +90,6 @@ import { selectHistoryPageFromCursor, selectRecentTimelineWindow, THREAD_HISTORY_PAGE_POLICY, - OLDER_THREAD_USER_TURN_LIMIT, } from "./threadHistoryPaging.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( @@ -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,18 +5121,13 @@ 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, + // The index pass already applied the requested user-turn limit, including find jumps. + maxUserTurns: undefined, maxItems: THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2), }, 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) => From b7c93d9b724f21ffe527a9051a58315059262375 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 16:26:19 +0000 Subject: [PATCH 6/7] test(server): cover notification context in SQL history pages --- .../src/orchestration-v2/ThreadFind.test.ts | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) 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; From e0fedfd6dd8e8b945031c8c075474c99d93d25a3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 16:30:03 +0000 Subject: [PATCH 7/7] fix(server): retain full turn context in SQL history pages --- .../src/orchestration-v2/ProjectionStore.ts | 48 +++++++++++-------- 1 file changed, 28 insertions(+), 20 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 098e10b7fca03..375755d4f3cb6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -88,8 +88,7 @@ import { threadHistoryTurnStarts, THREAD_HISTORY_MAX_RAW_TURNS, selectHistoryPageFromCursor, - selectRecentTimelineWindow, - THREAD_HISTORY_PAGE_POLICY, + projectedRowEncodedBytes, } from "./threadHistoryPaging.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( @@ -5072,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, @@ -5121,24 +5121,32 @@ export const layer: Layer.Layer = }; }), ); - 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, - // The index pass already applied the requested user-turn limit, including find jumps. - maxUserTurns: undefined, - 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, }; }), )