From 2f24391abd9088874540b31395ea35e4d158bc60 Mon Sep 17 00:00:00 2001 From: luke2x <97411838+luke2x@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:31:46 +0100 Subject: [PATCH 1/4] fix(history): enforce page budgets inside human turns and discard oversized caches --- .../orchestration-v2/ProjectionStore.test.ts | 51 +++++++- .../src/orchestration-v2/ProjectionStore.ts | 16 +-- .../threadHistoryPaging.test.ts | 115 ++++++++++++++---- .../orchestration-v2/threadHistoryPaging.ts | 16 +-- .../src/platform/orchestrationCache.test.ts | 14 ++- .../src/platform/orchestrationCache.ts | 3 +- .../src/state/threads-sync.test.ts | 65 ++++++++++ 7 files changed, 231 insertions(+), 49 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index fcadd8036b924..debeff58660da 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -40,6 +40,7 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import { buildBoundedThreadProjection, decodeThreadHistoryCursor, + encodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY, } from "./threadHistoryPaging.ts"; @@ -1529,6 +1530,52 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { [baselineId], ); } + const humanWindow = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }); + assert.lengthOf(humanWindow.projection.turnItems, 77); + + // Fewer than 75 rows still need a byte pass after materializing wire payloads. + yield* sql` + UPDATE orchestration_v2_projection_turn_items + SET type = 'assistant_message', payload_json = json_set(payload_json, + '$.type', 'assistant_message', '$.text', ${"x".repeat(40_000)}, + '$.messageId', turn_item_id, '$.streaming', json('false')) + WHERE thread_id = ${threadId} AND ordinal BETWEEN 2 AND 70 + `; + const byteCursor = encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: TurnItemId.make("turn-item:bounded-sql-history:71"), + position: 70, + }); + const bytePage = yield* projectionStore.getThreadHistoryPage(threadId, byteCursor); + assert.isAbove(bytePage.items.length, 0); + assert.isBelow(bytePage.items.length, 70); + assert.isAtMost( + Buffer.byteLength(JSON.stringify(bytePage.items), "utf8"), + THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes, + ); + const byteIds = bytePage.items.map((row) => String(row.sourceItemId)); + let byteNext = bytePage.nextCursor; + for (let pages = 0; byteNext !== null; pages += 1) { + assert.isBelow(pages, 70); + const page = yield* projectionStore.getThreadHistoryPage(threadId, byteNext); + assert.isAbove(page.items.length, 0); + byteIds.unshift(...page.items.map((row) => String(row.sourceItemId))); + byteNext = page.nextCursor; + } + assert.deepEqual( + byteIds, + Array.from({ length: 70 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`), + ); + yield* sql` + UPDATE orchestration_v2_projection_turn_items + SET type = 'command_execution', payload_json = json_set(payload_json, '$.type', 'command_execution') + WHERE thread_id = ${threadId} AND ordinal BETWEEN 2 AND 70 + `; + const retainedRequestPlan = yield* sql<{ readonly detail: string }>` EXPLAIN QUERY PLAN WITH selected AS ( @@ -1571,9 +1618,9 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const sqlPageLimit = THREAD_HISTORY_PAGE_POLICY.maxItems + 2; const initialSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, }); const initialPage = buildBoundedThreadProjection({ - policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined }, projection: initialSnapshot.projection, snapshotSequence: initialSnapshot.snapshotSequence, }); @@ -1586,10 +1633,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const anchorItemId = TurnItemId.make(decodeThreadHistoryCursor(cursor).si); const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, + userTurnLimit: 20, anchorItemId, }); const page = selectHistoryPageFromCursor({ - policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined }, items: snapshot.projection.visibleTurnItems, cursor, snapshotSequence: snapshot.snapshotSequence, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 00333529dd1fb..2f33111afdfd2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -2963,11 +2963,7 @@ export const layer: Layer.Layer = FROM eligible WHERE ordinal >= (SELECT ordinal FROM boundary) ORDER BY ordinal DESC, turn_item_id DESC - LIMIT CASE - WHEN ${window.rowLimit} = 0 THEN 0 - WHEN (SELECT COUNT(*) FROM user_anchors) > 0 THEN -1 - ELSE ${window.rowLimit} - END + LIMIT ${window.rowLimit} ), retained AS MATERIALIZED ( SELECT item_rowid, ordinal, turn_item_id FROM selected UNION @@ -5496,12 +5492,6 @@ export const layer: Layer.Layer = }; }), ); - if (hasUserTurns) - return { - ...page, - snapshotSequence, - items: items.map((row, position) => ({ ...row, position })), - }; const bounded = selectRecentTimelineWindow({ items, snapshotSequence, @@ -7382,7 +7372,9 @@ export const layerMemory: Layer.Layer = Layer.effect( ? rawStart : anchors.at(-anchorLimit)! : Math.max(0, anchorIndex - options.rowLimit); - const visibleTurnItems = candidates.slice(start); + const visibleTurnItems = candidates.slice( + Math.max(start, anchorIndex - Math.max(0, options.rowLimit)), + ); return { ...snapshot, projection: { ...snapshot.projection, visibleTurnItems }, diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 4055fd85b7ae5..9c926f272ce54 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -176,6 +176,53 @@ function makeProjection(visibleTurnItems: OrchestrationV2ProjectedTurnItem[]) { } describe("threadHistoryPaging", () => { + it("bounds a single human turn and pages every item exactly once", () => { + const items = Array.from({ length: 180 }, (_, index) => makeRow(index)); + items[0] = { + ...items[0]!, + item: { + ...items[0]!.item, + type: "user_message", + createdBy: "user", + creationSource: "web", + inputIntent: "turn_start", + messageId: MessageId.make("long-human-turn"), + text: "Continue", + attachments: [], + }, + } as OrchestrationV2ProjectedTurnItem; + const recent = selectRecentTimelineWindow({ items, snapshotSequence: 1 }); + expect(recent.items).toHaveLength(THREAD_HISTORY_PAGE_POLICY.maxItems); + const loaded = [...recent.items]; + let cursor = recent.nextCursor; + for (let pages = 0; cursor !== null; pages += 1) { + expect(pages).toBeLessThan(items.length); + const page = selectHistoryPageFromCursor({ items, cursor, snapshotSequence: 1 }); + expect(page.items.length).toBeGreaterThan(0); + expect(page.items.length).toBeLessThanOrEqual(THREAD_HISTORY_PAGE_POLICY.maxItems); + loaded.unshift(...page.items); + cursor = page.nextCursor; + } + expect(loaded.map((row) => row.sourceItemId)).toEqual(items.map((row) => row.sourceItemId)); + + const large = items + .slice(0, 12) + .map((row) => + row.item.type === "command_execution" + ? { ...row, item: { ...row.item, output: "x".repeat(200_000) } } + : row, + ); + const bounded = buildBoundedThreadProjection({ + projection: makeProjection(large), + snapshotSequence: 1, + }); + expect(bounded.projection.visibleTurnItems.length).toBeLessThan(large.length); + expect(bounded.payloadBudgetExceeded).toBe(false); + expect(boundedTimelineEncodedBytes(bounded.projection)).toBeLessThanOrEqual( + THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes, + ); + }); + 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); @@ -195,13 +242,15 @@ describe("threadHistoryPaging", () => { }; return [prompt, makeRow(turn * 2 + 1)]; }).flat(); - const first = selectRecentTimelineWindow({ items, snapshotSequence: 1 }); + const policy = { ...THREAD_HISTORY_PAGE_POLICY, 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); @@ -244,13 +293,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 = { ...THREAD_HISTORY_PAGE_POLICY, 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), @@ -288,26 +339,46 @@ describe("threadHistoryPaging", () => { }); it("always includes at least one pathological oversized item", () => { - const items = [makeRow(0, { outputBytes: 2_000_000 }), makeRow(1, { outputBytes: 10 })]; - const page = selectRecentTimelineWindow({ - items, - snapshotSequence: 1, - policy: { maxItems: 50, maxEncodedBytes: 1_024 }, - }); - expect(page.items).toHaveLength(1); - expect(page.items[0]?.sourceItemId).toBe("item-1"); - expect(page.hasMoreHistory).toBe(true); - - const older = selectHistoryPageFromCursor({ - items, - cursor: page.nextCursor!, - snapshotSequence: 1, - policy: { maxItems: 50, maxEncodedBytes: 1_024 }, - }); - expect(older.items).toHaveLength(1); - expect(older.items[0]?.sourceItemId).toBe("item-0"); - expect(older.hasMoreHistory).toBe(false); - expect(older.nextCursor).toBeNull(); + for (const humanTurn of [false, true]) { + const oversized = makeRow(0, { outputBytes: 2_000_000 }); + const items = [ + humanTurn + ? ({ + ...oversized, + item: { + ...oversized.item, + type: "user_message", + createdBy: "user", + creationSource: "web", + inputIntent: "turn_start", + messageId: MessageId.make("oversized-human-turn"), + text: "x".repeat(2_000_000), + attachments: [], + }, + } as OrchestrationV2ProjectedTurnItem) + : oversized, + makeRow(1, { outputBytes: 10 }), + ]; + const policy = { ...THREAD_HISTORY_PAGE_POLICY, maxItems: 50, maxEncodedBytes: 1_024 }; + const page = selectRecentTimelineWindow({ items, snapshotSequence: 1, policy }); + expect(page.items).toHaveLength(1); + expect(page.items[0]?.sourceItemId).toBe("item-1"); + expect(page.hasMoreHistory).toBe(true); + + const older = selectHistoryPageFromCursor({ + items, + cursor: page.nextCursor!, + snapshotSequence: 1, + policy, + }); + expect(older.items).toHaveLength(1); + expect(older.items[0]?.sourceItemId).toBe("item-0"); + expect(older.hasMoreHistory).toBe(false); + expect(older.nextCursor).toBeNull(); + expect([...older.items, ...page.items].map((row) => row.sourceItemId)).toEqual( + items.map((row) => row.sourceItemId), + ); + } }); it("recovers identity-miss cursors when position is past a shrunken timeline", () => { diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index e0aa215f3f1ef..2e114131c7a7e 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -7,8 +7,8 @@ import type { } from "@t3tools/contracts"; /** - * Item/byte budgets only apply to histories without turn starts; tool activity - * must not split an ordinary conversation turn. + * Bound every history window by rows and bytes, including a single long human + * turn. Turn limits can stop a window earlier. */ export const THREAD_HISTORY_PAGE_POLICY = { maxUserTurns: 10, @@ -210,8 +210,7 @@ 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. + * Always admit at least one row so oversized items cannot deadlock paging. */ function selectOlderTimelinePage(input: { readonly items: ReadonlyArray; @@ -238,12 +237,13 @@ function selectOlderTimelinePage(input: { : undefined; for (let index = end - 1; index >= 0; index -= 1) { const row = input.items[index]!; - const rowBytes = turnLimit === undefined ? rowCost(row) : 0; + const rowBytes = rowCost(row); if ( selected.length > 0 && - (turnLimit === undefined - ? selected.length >= policy.maxItems || encodedBytes + rowBytes > policy.maxEncodedBytes - : userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS) + (selected.length >= policy.maxItems || + encodedBytes + rowBytes > policy.maxEncodedBytes || + (turnLimit !== undefined && + (userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS))) ) { break; } diff --git a/packages/client-runtime/src/platform/orchestrationCache.test.ts b/packages/client-runtime/src/platform/orchestrationCache.test.ts index eee72d9cbb3de..5ea3f2ee5bda5 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.test.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.test.ts @@ -74,14 +74,20 @@ describe("orchestration cache envelopes", () => { expect(decoded.visibleTurnItems.find((entry) => entry.item.id === item.id)?.item).toEqual(item); }); - it("rejects thread snapshots saved before queued-run items were kept, but not shells", () => { - const staleThread = encodeStoredThreadSnapshotJson({ + it("rejects thread snapshots saved before human history was bounded, but not shells", () => { + const currentThread = encodeStoredThreadSnapshotJson({ schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId: v2ThreadId, snapshot: { snapshotSequence: 4, projection: v2Projection }, - }).replace(`"schemaVersion":${ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION}`, `"schemaVersion":3`); - expect(() => decodeStoredThreadSnapshotJson(staleThread)).toThrow(); + }); + for (const staleVersion of [3, 4]) { + const staleThread = currentThread.replace( + `"schemaVersion":${ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION}`, + `"schemaVersion":${staleVersion}`, + ); + expect(() => decodeStoredThreadSnapshotJson(staleThread)).toThrow(); + } const shell = encodeStoredShellSnapshot({ schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, environmentId, diff --git a/packages/client-runtime/src/platform/orchestrationCache.ts b/packages/client-runtime/src/platform/orchestrationCache.ts index f0b95d3bb603b..e7800e20bd33a 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.ts @@ -10,7 +10,8 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; export const ORCHESTRATION_CACHE_SCHEMA_VERSION = 3 as const; -export const ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION = 4 as const; +// Older caches may contain human turns that bypassed server history budgets. +export const ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION = 5 as const; export const StoredOrchestrationShellSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index b87ab18a50ae7..6a3ca13346bf6 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -20,6 +20,7 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; @@ -35,6 +36,11 @@ import { import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; +import { + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + StoredOrchestrationThreadSnapshot, + decodeOrDiscardOrchestrationCache, +} from "../platform/orchestrationCache.ts"; import * as RpcSession from "../rpc/session.ts"; import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; @@ -320,6 +326,65 @@ const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => { }; describe("EnvironmentThreads", () => { + it.effect("discards an oversized version-4 cache and requests a fresh bounded snapshot", () => + Effect.gen(function* () { + const item: OrchestrationV2TurnItem = { + id: TurnItemId.make("old-unbounded-cache-item"), + threadId: THREAD_ID, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: BASE_PROJECTION.updatedAt, + completedAt: BASE_PROJECTION.updatedAt, + updatedAt: BASE_PROJECTION.updatedAt, + type: "system_notice", + message: "x".repeat(1_100_000), + }; + const schema = Schema.fromJsonString(StoredOrchestrationThreadSnapshot); + const encoded = yield* Schema.encodeEffect(schema)({ + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + environmentId: TARGET.environmentId, + threadId: THREAD_ID, + snapshot: { + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + projection: { ...BASE_PROJECTION, turnItems: [item] }, + }, + }); + const stale = encoded.replace( + `"schemaVersion":${ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION}`, + '"schemaVersion":4', + ); + let discarded = false; + const h = yield* makeHarness({ + loadCached: decodeOrDiscardOrchestrationCache( + Schema.decodeUnknownEffect(schema)(stale).pipe( + Effect.map((stored) => Option.some(stored.snapshot)), + ), + Effect.sync(() => { + discarded = true; + }), + ), + httpSnapshot: { + _tag: "present", + snapshot: { snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, projection: BASE_PROJECTION }, + history: { historyCursor: "fresh-bounded-cursor", hasMoreHistory: true }, + }, + }); + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(discarded).toBe(true); + expect(yield* Ref.get(h.loaderCalls)).toBe(1); + expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION); + expect(state.history.historyCursor).toBe("fresh-bounded-cursor"); + expect(state.history.hasMoreHistory).toBe(true); + }), + ); + it.effect("loads the server thread when its local cache read fails", () => Effect.gen(function* () { const h = yield* makeHarness({ From 7f822da3f96b5fe371ee26197008c8b1c147e19d Mon Sep 17 00:00:00 2001 From: luke2x <97411838+luke2x@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:54:48 +0100 Subject: [PATCH 2/4] fix(history): keep targeted pages and interrupt requests intact after bounding The byte pass in getThreadHistoryPage re-bounded a targeted page with the default 20-turn limit, dropping a throughEntryId target that the first pass had reached. Both passes now share one policy from olderHistoryPagePolicy. Row and byte budgets can now split a human turn between an interrupt result and its request. A page whose results need an older request now carries that request, so the client does not hide the result as a legacy steer. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 61 ++++++++ .../src/orchestration-v2/ProjectionStore.ts | 136 ++++++++++-------- .../orchestration-v2/threadHistoryPaging.ts | 94 ++++++++---- 3 files changed, 202 insertions(+), 89 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index debeff58660da..26c20a6cb6a5a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1576,6 +1576,67 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { WHERE thread_id = ${threadId} AND ordinal BETWEEN 2 AND 70 `; + // A targeted page that needs more than 20 human turns must still reach its target + // after the byte pass re-bounds it. + yield* sql` + UPDATE orchestration_v2_projection_turn_items + SET type = 'user_message', payload_json = json_set(payload_json, + '$.type', 'user_message', '$.createdBy', 'user', '$.creationSource', 'web', + '$.inputIntent', 'turn_start', '$.messageId', turn_item_id, '$.text', 'Turn', + '$.attachments', json('[]')) + WHERE thread_id = ${threadId} AND ordinal BETWEEN 600 AND 624 + `; + const targetedPage = yield* projectionStore.getThreadHistoryPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: TurnItemId.make("turn-item:bounded-sql-history:625"), + position: 624, + }), + "turn-item:bounded-sql-history:600", + ); + assert.strictEqual( + String(targetedPage.items[0]?.sourceItemId), + "turn-item:bounded-sql-history:600", + ); + yield* sql` + UPDATE orchestration_v2_projection_turn_items + SET type = 'command_execution', payload_json = json_set(payload_json, '$.type', 'command_execution') + WHERE thread_id = ${threadId} AND ordinal BETWEEN 600 AND 624 + `; + + // A turn split by the row budget keeps an interrupt result's request on the same page. + yield* sql` + UPDATE orchestration_v2_projection_turn_items + SET run_id = 'run:bounded-sql-history:500', + type = CASE ordinal WHEN 500 THEN 'run_interrupt_request' ELSE 'run_interrupt_result' END, + payload_json = json_set(payload_json, + '$.runId', 'run:bounded-sql-history:500', '$.message', 'Stop', + '$.type', CASE ordinal WHEN 500 THEN 'run_interrupt_request' ELSE 'run_interrupt_result' END) + WHERE thread_id = ${threadId} AND ordinal IN (500, 580) + `; + const interruptPage = yield* projectionStore.getThreadHistoryPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: TurnItemId.make("turn-item:bounded-sql-history:590"), + position: 589, + }), + ); + const interruptIds = interruptPage.items.map((row) => String(row.sourceItemId)); + assert.include(interruptIds, "turn-item:bounded-sql-history:580"); + assert.include(interruptIds, "turn-item:bounded-sql-history:500"); + assert.notInclude(interruptIds, "turn-item:bounded-sql-history:501"); + yield* sql` + UPDATE orchestration_v2_projection_turn_items + SET run_id = 'run:bounded-sql-history:' || ordinal, type = 'command_execution', + payload_json = json_set(payload_json, + '$.runId', 'run:bounded-sql-history:' || ordinal, '$.type', 'command_execution') + WHERE thread_id = ${threadId} AND ordinal IN (500, 580) + `; + const retainedRequestPlan = yield* sql<{ readonly detail: string }>` EXPLAIN QUERY PLAN WITH selected AS ( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2f33111afdfd2..6775a4a4bbc1e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -94,6 +94,8 @@ import { isConversationHistoryItem, threadHistoryTurnStarts, THREAD_HISTORY_MAX_RAW_TURNS, + missingInterruptRequests, + olderHistoryPagePolicy, selectHistoryPageFromCursor, selectRecentTimelineWindow, THREAD_HISTORY_PAGE_POLICY, @@ -5435,78 +5437,85 @@ 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 page = yield* Effect.try({ - try: () => - selectHistoryPageFromCursor({ - items: index.visible.map((row, position) => ({ - ...row, - position, - canonicalPosition: position, - })), + const indexed = index.visible.map((row, position) => ({ + ...row, + position, + canonicalPosition: position, + })); + const selection = yield* Effect.try({ + try: () => { + const policy = olderHistoryPagePolicy({ items: indexed, cursor, throughEntryId }); + const page = selectHistoryPageFromCursor({ + items: indexed, cursor, snapshotSequence, rowEncodedBytes: () => 0, - throughEntryId, - }), + policy, + }); + return { policy, page }; + }, catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), }); + const { policy, page } = selection; 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 - .filter((row) => !row.synthetic) - .map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })), - ); - const payloads = yield* sql<{ - thread_id: string; - turn_item_id: string; - payload_json: string; - }>` + const materialize = Effect.fn("ProjectionStore.materializeHistoryRows")(function* ( + rows: ReadonlyArray<(typeof indexed)[number]>, + ) { + if (rows.length === 0) return []; + const sources = yield* encodeTimelineSources( + rows + .filter((row) => !row.synthetic) + .map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })), + ); + const payloads = yield* sql<{ + thread_id: string; + turn_item_id: string; + payload_json: string; + }>` SELECT item.thread_id, item.turn_item_id, item.payload_json FROM orchestration_v2_projection_turn_items item JOIN json_each(${sources}) wanted ON item.thread_id = json_extract(wanted.value, '$.threadId') AND item.turn_item_id = json_extract(wanted.value, '$.id')`; - const byIdentity = new Map( - payloads.map((row) => [`${row.thread_id}:${row.turn_item_id}`, row.payload_json]), - ); - const items = yield* Effect.forEach(materialized, (row) => - Effect.gen(function* () { - const payload = byIdentity.get(`${row.sourceThreadId}:${row.sourceItemId}`); - const item = - row.synthetic ?? - (payload === undefined - ? yield* new ProjectionStoreReadError({ threadId }) - : yield* decodeTurnItemPayload(payload)); - return { - position: row.canonicalPosition, - visibility: row.visibility, - sourceThreadId: row.sourceThreadId, - sourceItemId: row.sourceItemId, - item: projectTurnItemForWire(item), - }; - }), - ); - 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 byIdentity = new Map( + payloads.map((row) => [`${row.thread_id}:${row.turn_item_id}`, row.payload_json]), + ); + return yield* Effect.forEach(rows, (row) => + Effect.gen(function* () { + const payload = byIdentity.get(`${row.sourceThreadId}:${row.sourceItemId}`); + const item = + row.synthetic ?? + (payload === undefined + ? yield* new ProjectionStoreReadError({ threadId }) + : yield* decodeTurnItemPayload(payload)); + return { + position: row.canonicalPosition, + visibility: row.visibility, + sourceThreadId: row.sourceThreadId, + sourceItemId: row.sourceItemId, + item: projectTurnItemForWire(item), + }; + }), + ); }); + const items = yield* materialize( + conversationOnly && hasUserTurns + ? page.items.filter((row) => isConversationHistoryItem(row.item)) + : page.items, + ); + const bounded = selectRecentTimelineWindow({ items, snapshotSequence, policy }); + const requests = yield* materialize(missingInterruptRequests(indexed, bounded.items)); + const pageItems = [...requests, ...bounded.items].map((row, position) => ({ + ...row, + position, + })); return { snapshotSequence, items: conversationOnly - ? bounded.items.filter((row) => isConversationHistoryItem(row.item)) - : bounded.items, + ? pageItems.filter((row) => isConversationHistoryItem(row.item)) + : pageItems, nextCursor: bounded.hasMoreHistory ? bounded.nextCursor : page.nextCursor, hasMoreHistory: bounded.hasMoreHistory || page.hasMoreHistory, }; @@ -7307,21 +7316,26 @@ export const layerMemory: Layer.Layer = Layer.effect( return yield* new ProjectionStoreThreadNotFoundError({ threadId }); return yield* Effect.try({ try: () => { + const visible = projection.visibleTurnItems.map((row) => ({ + ...row, + item: projectTurnItemForWire(row.item), + })); const page = selectHistoryPageFromCursor({ - items: projection.visibleTurnItems.map((row) => ({ - ...row, - item: projectTurnItemForWire(row.item), - })), + items: visible, cursor, throughEntryId, snapshotSequence, }); + const items = [ + ...missingInterruptRequests(visible, page.items), + ...page.items, + ].map((row, position) => ({ ...row, position })); return { ...page, snapshotSequence, items: conversationOnly - ? page.items.filter((row) => isConversationHistoryItem(row.item)) - : page.items, + ? items.filter((row) => isConversationHistoryItem(row.item)) + : items, }; }, catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 2e114131c7a7e..7253addf396ef 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -307,6 +307,43 @@ function findCursorIndex( return Math.min(cursor.p, items.length); } +/** + * Older pages allow twice the turns, and a targeted page widens further so it + * reaches `throughEntryId`. Callers that re-bound a page must reuse this policy + * or they drop the target again. + */ +export function olderHistoryPagePolicy(input: { + readonly items: ReadonlyArray; + readonly cursor: string; + readonly policy?: ThreadHistoryPagePolicy | undefined; + readonly throughEntryId?: string | undefined; +}): ThreadHistoryPagePolicy { + const policy = input.policy ?? { + ...THREAD_HISTORY_PAGE_POLICY, + maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, + }; + if (input.throughEntryId === undefined) return policy; + const anchorIndex = findCursorIndex(input.items, decodeThreadHistoryCursor(input.cursor)); + const targetIndex = input.items.findIndex( + (row) => (row.item.messageId ?? row.item.id) === input.throughEntryId, + ); + if (targetIndex < 0 || targetIndex >= anchorIndex) return policy; + const turns = input.items + .slice(targetIndex, anchorIndex) + .filter((row) => isThreadHistoryUserTurn(row.item)).length; + return { + ...policy, + maxUserTurns: Math.min( + OLDER_THREAD_USER_TURN_LIMIT * 2, + Math.max( + OLDER_THREAD_USER_TURN_LIMIT, + turns + (isThreadHistoryUserTurn(input.items[targetIndex]!.item) ? 0 : 1), + ), + ), + maxItems: policy.maxItems * 2, + }; +} + export function selectHistoryPageFromCursor(input: { readonly items: ReadonlyArray; readonly cursor: string; @@ -316,41 +353,42 @@ export function selectHistoryPageFromCursor(input: { readonly throughEntryId?: string | undefined; }): SelectTimelinePageResult { const decoded = decodeThreadHistoryCursor(input.cursor); - const anchorIndex = findCursorIndex(input.items, decoded); - let policy = input.policy ?? { - ...THREAD_HISTORY_PAGE_POLICY, - maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, - }; - if (input.throughEntryId !== undefined) { - const targetIndex = input.items.findIndex( - (row) => (row.item.messageId ?? row.item.id) === input.throughEntryId, - ); - if (targetIndex >= 0 && targetIndex < anchorIndex) { - const turns = input.items - .slice(targetIndex, anchorIndex) - .filter((row) => isThreadHistoryUserTurn(row.item)).length; - policy = { - ...policy, - maxUserTurns: Math.min( - OLDER_THREAD_USER_TURN_LIMIT * 2, - Math.max( - OLDER_THREAD_USER_TURN_LIMIT, - turns + (isThreadHistoryUserTurn(input.items[targetIndex]!.item) ? 0 : 1), - ), - ), - maxItems: policy.maxItems * 2, - }; - } - } return selectOlderTimelinePage({ items: input.items, - exclusiveEndIndex: anchorIndex, + exclusiveEndIndex: findCursorIndex(input.items, decoded), snapshotSequence: input.snapshotSequence, ...(input.rowEncodedBytes === undefined ? {} : { rowEncodedBytes: input.rowEncodedBytes }), - policy, + policy: olderHistoryPagePolicy(input), }); } +/** + * Row and byte budgets can split a turn between an interrupt result and its + * older request. Clients read a superseded result without a request as a + * legacy steer and hide it, so a page carries the requests its results need. + */ +export function missingInterruptRequests( + items: ReadonlyArray, + page: ReadonlyArray>, +): Row[] { + const runs = new Set(); + for (const row of page) { + if (row.item.type === "run_interrupt_result" && row.item.runId != null) + runs.add(row.item.runId); + } + for (const row of page) { + if (row.item.type === "run_interrupt_request" && row.item.runId != null) + runs.delete(row.item.runId); + } + if (runs.size === 0) return []; + return items.filter( + (row) => + row.item.type === "run_interrupt_request" && + row.item.runId != null && + runs.has(row.item.runId), + ); +} + function isLocalProjectedRow( projection: OrchestrationV2ThreadProjection, row: OrchestrationV2ProjectedTurnItem, From 56e2a880a84f815fa8173187f2033ff0fb9d0e5e Mon Sep 17 00:00:00 2001 From: luke2x <97411838+luke2x@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:32:45 +0100 Subject: [PATCH 3/4] test(history): expect long turns to page by size ThreadFind still required every page to start at a user message, the whole-turn rule this PR removes. It now checks that each page fits the byte budget (or holds one oversized item) and that every message is paged exactly once. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ThreadFind.test.ts | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index 3d261ed85e2f2..6793421a64333 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -24,7 +24,11 @@ import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as EventStore from "./EventStore.ts"; -import { encodeThreadHistoryCursor, selectHistoryPageFromCursor } from "./threadHistoryPaging.ts"; +import { + encodeThreadHistoryCursor, + selectHistoryPageFromCursor, + THREAD_HISTORY_PAGE_POLICY, +} from "./threadHistoryPaging.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; const layerTest = Layer.mergeAll(EventStore.layer, ProjectionStore.layer, ProjectStore.layer).pipe( @@ -722,7 +726,7 @@ describe("V2 history materialization", () => { }).pipe(Effect.provide(layerTest)), ); - it.effect("pages complete turns without duplicating or skipping messages", () => + it.effect("pages long turns without duplicating or skipping messages", () => Effect.gen(function* () { const projection = yield* setup; const items = Array.from({ length: 45 }, (_, turn) => [ @@ -763,7 +767,12 @@ describe("V2 history materialization", () => { snapshotSequence: snapshot.snapshotSequence, }), }); - assert.equal(page.items[0]?.item.type, "user_message"); + // A turn larger than the byte budget is split, but only a lone oversized item may exceed it. + assert.isTrue( + page.items.length === 1 || + Buffer.byteLength(JSON.stringify(page.items), "utf8") <= + THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes, + ); collected.unshift(...page.items.map((row) => String(row.sourceItemId))); cursor = page.nextCursor; } From c939917e99a750301379c2b495310004aa9a588c Mon Sep 17 00:00:00 2001 From: luke2x <97411838+luke2x@users.noreply.github.com> Date: Sun, 25 Oct 2026 01:33:26 +0100 Subject: [PATCH 4/4] fix(history): honor stricter maxTurnItems on human-turn pages Human-turn windows now use the smaller of maxItems and maxTurnItems, so a caller that sets a stricter maxTurnItems is respected while maxItems remains the upper bound. Adds a regression test for the stricter limit. --- .../threadHistoryPaging.test.ts | 32 ++++++++++++++++++- .../orchestration-v2/threadHistoryPaging.ts | 8 +++-- 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 2125e2985f1d0..b2af7e6728256 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -201,6 +201,31 @@ describe("threadHistoryPaging", () => { ); }); + it("honors a stricter maxTurnItems on human-turn windows", () => { + const items = Array.from({ length: 60 }, (_, index) => makeRow(index)); + items[0] = { + ...items[0]!, + item: { + ...items[0]!.item, + type: "user_message", + createdBy: "user", + creationSource: "web", + inputIntent: "turn_start", + messageId: MessageId.make("stricter-turn-limit"), + text: "Continue", + attachments: [], + }, + } as OrchestrationV2ProjectedTurnItem; + + const recent = selectRecentTimelineWindow({ + items, + snapshotSequence: 1, + policy: { ...THREAD_HISTORY_PAGE_POLICY, maxItems: 100, maxTurnItems: 20 }, + }); + + expect(recent.items).toHaveLength(20); + }); + 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); @@ -271,7 +296,12 @@ describe("threadHistoryPaging", () => { inRun(run * 4 + 2, run, notification(`Steer ${run}`)), inRun(run * 4 + 3, run), ]).flat(); - const policy = { ...THREAD_HISTORY_PAGE_POLICY, maxItems: 1_000, maxEncodedBytes: 10_000_000 }; + const policy = { + ...THREAD_HISTORY_PAGE_POLICY, + maxItems: 1_000, + maxTurnItems: 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}`); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index da0ca78522753..a0fbbe99d4a05 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -218,12 +218,16 @@ function selectOlderTimelinePage(input: { const turnLimit = candidates.some((row) => isThreadHistoryUserTurn(row.item)) ? policy.maxUserTurns : undefined; + const maxItems = + turnLimit === undefined + ? policy.maxItems + : Math.min(policy.maxItems, policy.maxTurnItems ?? policy.maxItems); 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 || + (selected.length >= maxItems || encodedBytes + rowBytes > policy.maxEncodedBytes || (turnLimit !== undefined && (userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS))) @@ -238,7 +242,7 @@ function selectOlderTimelinePage(input: { // Inside a turn longer than the window, no selected row starts a user turn. // Later pages and the snapshot query see only such rows and use `maxItems`, // so this page uses it too. - if (userTurns === 0) selected.splice(policy.maxItems); + if (userTurns === 0) selected.splice(maxItems); selected.reverse(); const oldest = selected[0];