From cc2cd48e4fa0e57e5d92e3052e5e4d77b3fff02a 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/5] 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 14fed37c0618..ef8b5e18afaa 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 438cf79138dd..b90541e6cd51 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3000,11 +3000,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 @@ -5535,12 +5531,6 @@ export const layer: Layer.Layer = }; }), ); - if (hasUserTurns) - return { - ...page, - snapshotSequence, - items: items.map((row, position) => ({ ...row, position })), - }; const bounded = selectRecentTimelineWindow({ items, snapshotSequence, @@ -7448,7 +7438,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 4055fd85b7ae..9c926f272ce5 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 e0aa215f3f1e..2e114131c7a7 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 eee72d9cbb3d..5ea3f2ee5bda 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 f0b95d3bb603..e7800e20bd33 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 a772eaaf76c2..37dd30c73233 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"; @@ -36,6 +37,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 { EMPTY_THREAD_HISTORY_META } from "./threadHistoryMerge.ts"; @@ -328,6 +334,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 ec2fc24d99e32fc4129be7223d76a3e55c0a8142 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/5] 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 ef8b5e18afaa..7a2caf841b50 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 b90541e6cd51..4735a040e50d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -96,6 +96,8 @@ import { isConversationHistoryItem, threadHistoryTurnStarts, THREAD_HISTORY_MAX_RAW_TURNS, + missingInterruptRequests, + olderHistoryPagePolicy, selectHistoryPageFromCursor, selectRecentTimelineWindow, THREAD_HISTORY_PAGE_POLICY, @@ -5474,78 +5476,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, }; @@ -7373,21 +7382,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 2e114131c7a7..7253addf396e 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 49984f0768e9cda458e4908f9eec549b63a65d30 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/5] 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 3d261ed85e2f..6793421a6433 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 a921e628005e5ac3a8e41d22436e270807e3c797 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 24 Oct 2026 16:05:28 -0700 Subject: [PATCH 4/5] fix(server): keep user turns whole in thread windows up to 500 rows A window that holds a user turn now stops at 10 user turns (20 on older pages), 500 rows, or the byte budget. Inside one turn longer than 500 rows, the window pages with the 75-row budget, and the snapshot query applies the same rule before it decodes payloads. This keeps ordinary threads whole, as V1's 500-activity cap did, while long agent turns stay bounded. The old-client getThreadProjection path now strips tool bodies before it applies the byte budget. Bounded snapshots no longer compute or send payloadBudgetExceeded: no client read it, and it cost an extra encode. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 30 ++++-- .../src/orchestration-v2/ProjectionStore.ts | 30 +++++- .../src/orchestration-v2/ThreadStream.test.ts | 1 - .../src/orchestration-v2/ThreadStream.ts | 1 - .../threadHistoryPaging.test.ts | 99 +++++++------------ .../orchestration-v2/threadHistoryPaging.ts | 52 ++++------ apps/server/src/ws.ts | 7 +- .../state/boundedThreadSnapshotHttp.test.ts | 1 - .../src/state/threads-atoms.test.ts | 1 - .../src/state/threads-sync.test.ts | 3 - packages/contracts/src/orchestrationV2.ts | 2 - 11 files changed, 108 insertions(+), 119 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 7a2caf841b50..419e4f9d4eb2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -43,6 +43,7 @@ import { encodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY, + THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, } from "./threadHistoryPaging.ts"; const layerTest = Layer.mergeAll( @@ -1530,11 +1531,24 @@ 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); + const humanWindow = () => + projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }); + assert.lengthOf( + (yield* humanWindow()).projection.turnItems, + THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, + ); + // With the newest user turn 1,000 rows back, the window pages from inside it. + const setNewestTurnType = (type: "command_execution" | "user_message") => sql` + UPDATE orchestration_v2_projection_turn_items + SET type = ${type}, payload_json = json_set(payload_json, '$.type', ${type}) + WHERE thread_id = ${threadId} AND ordinal = 1000 + `; + yield* setNewestTurnType("command_execution"); + assert.lengthOf((yield* humanWindow()).projection.turnItems, 77); + yield* setNewestTurnType("user_message"); // Fewer than 75 rows still need a byte pass after materializing wire payloads. yield* sql` @@ -1697,6 +1711,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { userTurnLimit: 20, anchorItemId, }); + assert.isAtMost( + snapshot.projection.visibleTurnItems.length, + THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, + ); const page = selectHistoryPageFromCursor({ items: snapshot.projection.visibleTurnItems, cursor, @@ -1707,7 +1725,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { cursor = page.nextCursor; } - assert.isAtLeast(pageCount, 4); + assert.isAtLeast(pageCount, 3); assert.lengthOf(loadedIds, 1_000); assert.strictEqual(new Set(loadedIds).size, 1_000); assert.deepEqual( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4735a040e50d..4ddc34ae4a4f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -101,7 +101,7 @@ import { selectHistoryPageFromCursor, selectRecentTimelineWindow, THREAD_HISTORY_PAGE_POLICY, - OLDER_THREAD_USER_TURN_LIMIT, + THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, } from "./threadHistoryPaging.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( @@ -2994,6 +2994,17 @@ export const layer: Layer.Layer = ELSE 0 END AS ordinal FROM user_anchors + ), latest_user_turn AS ( + -- Mirrors selectOlderTimelinePage: whole turns while the + -- newest user turn fits the turn row budget, otherwise the + -- window pages from inside that turn with the row limit. + SELECT COUNT(*) AS row_count + FROM ( + SELECT 1 + FROM eligible + WHERE ordinal >= (SELECT MAX(ordinal) FROM user_anchors) + LIMIT ${THREAD_HISTORY_PAGE_POLICY.maxTurnItems + 1} + ) ), selected AS ( -- Choose and sort rows by ID, then fetch payloads in that -- order. A turn window can hold megabytes of tool output, and @@ -3002,7 +3013,13 @@ export const layer: Layer.Layer = FROM eligible WHERE ordinal >= (SELECT ordinal FROM boundary) ORDER BY ordinal DESC, turn_item_id DESC - LIMIT ${window.rowLimit} + LIMIT CASE + WHEN ${window.rowLimit} = 0 THEN 0 + WHEN (SELECT row_count FROM latest_user_turn) + BETWEEN 1 AND ${THREAD_HISTORY_PAGE_POLICY.maxTurnItems} + THEN ${THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT} + ELSE ${window.rowLimit} + END ), retained AS MATERIALIZED ( SELECT item_rowid, ordinal, turn_item_id FROM selected UNION @@ -7452,8 +7469,15 @@ export const layerMemory: Layer.Layer = Layer.effect( ? rawStart : anchors.at(-anchorLimit)! : Math.max(0, anchorIndex - options.rowLimit); + const latestTurnRows = anchors.length > 0 ? anchorIndex - anchors.at(-1)! : 0; + const rowLimit = + options.rowLimit > 0 && + latestTurnRows > 0 && + latestTurnRows <= THREAD_HISTORY_PAGE_POLICY.maxTurnItems + ? THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT + : options.rowLimit; const visibleTurnItems = candidates.slice( - Math.max(start, anchorIndex - Math.max(0, options.rowLimit)), + Math.max(start, anchorIndex - Math.max(0, rowLimit)), ); return { ...snapshot, diff --git a/apps/server/src/orchestration-v2/ThreadStream.test.ts b/apps/server/src/orchestration-v2/ThreadStream.test.ts index 438a24743a73..a5059eada443 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.test.ts @@ -161,7 +161,6 @@ describe("decideThreadResume", () => { expect(snapshot.historyCursor).not.toBeNull(); expect(snapshot.hasMoreHistory).toBe(true); expect(snapshot.latestLocalTurnOrdinal).toBe(80); - expect(snapshot.payloadBudgetExceeded).toBe(false); }); it("rejects a 10 MiB raw replay even when its projected form fits the wire budget", () => { diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index 4ef21edd3b3c..7983093a3b9e 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -52,7 +52,6 @@ export function boundedSnapshotResponseFields(input: { historyCursor: bounded.historyCursor, hasMoreHistory: bounded.hasMoreHistory, latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal, - payloadBudgetExceeded: bounded.payloadBudgetExceeded, ...(compactProjection === null ? {} : { turnItemsOmitLocalVisible: true as const }), }; } diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 9c926f272ce5..38ad3b5c7de1 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -106,27 +106,6 @@ function interruptRow( }; } -function assistantRow( - text: string, - visibility: "local" | "inherited" = "local", -): OrchestrationV2ProjectedTurnItem { - const row = makeRow(0); - const sourceThreadId = visibility === "local" ? THREAD : ThreadId.make("parent-thread"); - return { - ...row, - visibility, - sourceThreadId, - item: { - ...row.item, - threadId: sourceThreadId, - type: "assistant_message", - messageId: MessageId.make("answer-0"), - text, - streaming: false, - }, - } as OrchestrationV2ProjectedTurnItem; -} - function makeProjection(visibleTurnItems: OrchestrationV2ProjectedTurnItem[]) { return { thread: { @@ -176,21 +155,30 @@ 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; + it("keeps a human turn whole up to the turn row budget and pages a longer one from inside", () => { + const humanTurn = (length: number) => { + const items = Array.from({ length }, (_, 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; + return items; + }; + + const whole = selectRecentTimelineWindow({ items: humanTurn(180), snapshotSequence: 1 }); + expect(whole.items).toHaveLength(180); + expect(whole.hasMoreHistory).toBe(false); + + const items = humanTurn(THREAD_HISTORY_PAGE_POLICY.maxTurnItems + 100); const recent = selectRecentTimelineWindow({ items, snapshotSequence: 1 }); expect(recent.items).toHaveLength(THREAD_HISTORY_PAGE_POLICY.maxItems); const loaded = [...recent.items]; @@ -199,7 +187,11 @@ describe("threadHistoryPaging", () => { 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); + expect(page.items.length).toBeLessThanOrEqual( + page.items[0]?.item.type === "user_message" + ? THREAD_HISTORY_PAGE_POLICY.maxTurnItems + : THREAD_HISTORY_PAGE_POLICY.maxItems, + ); loaded.unshift(...page.items); cursor = page.nextCursor; } @@ -217,7 +209,6 @@ describe("threadHistoryPaging", () => { 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, ); @@ -293,7 +284,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}`); @@ -628,7 +624,6 @@ describe("threadHistoryPaging", () => { const serializedBytes = Buffer.byteLength(JSON.stringify(bounded.projection), "utf8"); expect(serializedBytes).toBeLessThanOrEqual(THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes); - expect(bounded.payloadBudgetExceeded).toBe(false); expect(bounded.projection.plans).toEqual([]); expect(bounded.projection.contextHandoffs).toEqual([]); expect(populated.plans).toHaveLength(120); @@ -637,7 +632,7 @@ describe("threadHistoryPaging", () => { ).toBe(2_097_152); }); - it("preserves oversized actionable state and reports the budget exception", () => { + it("preserves oversized actionable state", () => { const projection = makeProjection([makeRow(0)]); const actionable = { ...projection, @@ -656,30 +651,10 @@ describe("threadHistoryPaging", () => { const bounded = buildBoundedThreadProjection({ projection: actionable, snapshotSequence: 9 }); - expect(bounded.payloadBudgetExceeded).toBe(true); expect(bounded.projection.plans[0]).toEqual(actionable.plans[0]); expect(bounded.projection.visibleTurnItems).toEqual(actionable.visibleTurnItems); }); - it("flips overflow at the exact encoded size when timeline text fits the budget", () => { - // Inherited rows are encoded once, local rows twice; neither may overflow early. - for (const visibility of ["local", "inherited"] as const) { - const projection = makeProjection([assistantRow("a".repeat(20_000), visibility)]); - const build = (maxEncodedBytes: number) => - buildBoundedThreadProjection({ - projection, - snapshotSequence: 1, - policy: { maxItems: 75, maxEncodedBytes }, - }); - const bytes = Buffer.byteLength( - JSON.stringify(build(Number.MAX_SAFE_INTEGER).projection), - "utf8", - ); - expect(build(bytes).payloadBudgetExceeded).toBe(false); - expect(build(bytes - 1).payloadBudgetExceeded).toBe(true); - } - }); - it("keeps agent-only turns visible when checkpoint file lists fill the byte budget", () => { const items = Array.from({ length: 3 }, (_, turn) => { const first = makeRow(turn * 5); @@ -712,7 +687,6 @@ describe("threadHistoryPaging", () => { expect(bounded.projection.visibleTurnItems).toHaveLength(items.length); expect(bounded.hasMoreHistory).toBe(false); - expect(bounded.payloadBudgetExceeded).toBe(true); }); it("keeps paged historical plan detail in the turn item and only status in its artifact", () => { @@ -756,6 +730,5 @@ describe("threadHistoryPaging", () => { markdown: "", detailInTurnItem: true, }); - expect(bounded.payloadBudgetExceeded).toBe(false); }); }); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 7253addf396e..dee44adacbb2 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -7,12 +7,15 @@ import type { } from "@t3tools/contracts"; /** - * Bound every history window by rows and bytes, including a single long human - * turn. Turn limits can stop a window earlier. + * Every window stops at the byte budget. A window that holds a user turn also + * stops at `maxUserTurns` or `maxTurnItems` rows, which keeps ordinary turns + * whole (V1 kept 500 activities). Any other window, such as an agent-driven + * thread or the inside of one very long turn, stops at `maxItems` rows. */ export const THREAD_HISTORY_PAGE_POLICY = { maxUserTurns: 10, maxItems: 75, + maxTurnItems: 500, maxEncodedBytes: 1_048_576, } as const; @@ -24,12 +27,16 @@ export function isConversationHistoryItem(item: Pick { - if ("text" in item) { - length += item.text.length; - } - return length > budget; - }; - return ( - projection.visibleTurnItems.some((row) => exceeds(row.item)) || - projection.turnItems.some(exceeds) - ); -} - export function projectedRowEncodedBytes(row: HistoryRow): number { return bytesOfJson(row); } @@ -235,12 +220,14 @@ function selectOlderTimelinePage(input: { const turnLimit = candidates.some((row) => isThreadHistoryUserTurn(row.item)) ? policy.maxUserTurns : undefined; + const maxItems = + turnLimit === undefined ? 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))) @@ -252,6 +239,10 @@ function selectOlderTimelinePage(input: { if (isThreadHistoryUserTurn(row.item)) userTurns += 1; if (turnStarts[index]) rawTurns += 1; } + // 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); selected.reverse(); const oldest = selected[0]; @@ -341,6 +332,7 @@ export function olderHistoryPagePolicy(input: { ), ), maxItems: policy.maxItems * 2, + maxTurnItems: (policy.maxTurnItems ?? policy.maxItems) * 2, }; } @@ -567,11 +559,7 @@ export function buildBoundedThreadProjection(input: { policy.maxEncodedBytes / 2, policy.maxEncodedBytes - dependencyReserve - controlBytes - 1_024, ); - const windowPolicy: ThreadHistoryPagePolicy = { - maxUserTurns: policy.maxUserTurns, - maxItems: policy.maxItems, - maxEncodedBytes: windowBudget, - }; + const windowPolicy: ThreadHistoryPagePolicy = { ...policy, maxEncodedBytes: windowBudget }; const window = selectRecentTimelineWindow({ items: controlProjection.visibleTurnItems, @@ -642,9 +630,5 @@ export function buildBoundedThreadProjection(input: { historyCursor: window.nextCursor, hasMoreHistory: window.hasMoreHistory, latestLocalTurnOrdinal, - payloadBudgetExceeded: - controlBytes > policy.maxEncodedBytes || - timelineTextExceedsBudget(projection, policy.maxEncodedBytes) || - bytesOfJson(projection) > policy.maxEncodedBytes, }; } diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 9a2dcc71b81b..b8113e7ca400 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1968,13 +1968,12 @@ export const layerWsRpc = ( rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, }) .pipe( - Effect.map((snapshot) => - projectThreadProjectionForWire( + Effect.map( + (snapshot) => buildBoundedThreadProjection({ - projection: snapshot.projection, + projection: projectThreadProjectionForWire(snapshot.projection), snapshotSequence: snapshot.snapshotSequence, }).projection, - ), ), Effect.mapError( (cause) => diff --git a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts index 5caaab1459ca..52ce11411af7 100644 --- a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts +++ b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts @@ -115,7 +115,6 @@ const boundedBody = (compact: boolean) => ({ historyCursor: "bounded-cursor", hasMoreHistory: true, latestLocalTurnOrdinal: 2, - payloadBudgetExceeded: false, ...(compact ? { turnItemsOmitLocalVisible: true } : {}), }); diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 01308c26d86d..b086612aa451 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -339,7 +339,6 @@ describe("createEnvironmentThreadStateAtoms", () => { historyCursor: "socket-older-1", hasMoreHistory: true, latestLocalTurnOrdinal: 4, - payloadBudgetExceeded: false, }); yield* Queue.offer(subscription.events, { kind: "synchronized" }); yield* observeState( diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 37dd30c73233..9956694aa193 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -1149,7 +1149,6 @@ describe("EnvironmentThreads", () => { historyCursor: "socket-history-cursor", hasMoreHistory: true, latestLocalTurnOrdinal: 47, - payloadBudgetExceeded: false, }); const state = yield* awaitThreadState( @@ -1264,7 +1263,6 @@ describe("EnvironmentThreads", () => { historyCursor: "progressive-cursor", hasMoreHistory: true, latestLocalTurnOrdinal: 47, - payloadBudgetExceeded: false, }); yield* awaitThreadState( harness.observed, @@ -1373,7 +1371,6 @@ describe("EnvironmentThreads", () => { historyCursor: "compact-cursor", hasMoreHistory: true, latestLocalTurnOrdinal: 2, - payloadBudgetExceeded: false, turnItemsOmitLocalVisible: true, }); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index a8d21845ad5f..b708f3187d1a 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3324,7 +3324,6 @@ export const OrchestrationV2ThreadBoundedSnapshot = Schema.Struct({ historyCursor: Schema.NullOr(TrimmedNonEmptyString), hasMoreHistory: Schema.Boolean, latestLocalTurnOrdinal: Schema.NullOr(NonNegativeInt), - payloadBudgetExceeded: Schema.optional(Schema.Boolean), /** * Set only for clients that opted in: `projection.turnItems` omits the items * of local visible rows, which lead the full list. Clients must restore them @@ -3408,7 +3407,6 @@ export const OrchestrationV2ThreadStreamItem = Schema.Union([ historyCursor: Schema.optionalKey(Schema.NullOr(TrimmedNonEmptyString)), hasMoreHistory: Schema.optionalKey(Schema.Boolean), latestLocalTurnOrdinal: Schema.optionalKey(Schema.NullOr(NonNegativeInt)), - payloadBudgetExceeded: Schema.optionalKey(Schema.Boolean), turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)), }), Schema.Struct({ From 944fe094a8c9c15d1e989f010c6efbd411b3ed2e Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 24 Oct 2026 16:05:30 -0700 Subject: [PATCH 5/5] test(server): cap the compact cold open of a long agent turn The transfer budget test now opens a thread whose last user turn holds 900 rows of reasoning, commands, file changes, MCP calls, progress messages, checkpoints, and subagents. It loads the compact bounded snapshot that web and mobile use and one older page, and caps decoded bytes and rows. On main the cold open carried all 920 rows (1.07 MB decoded). The internal performance doc now describes the real window rules and the bodies that still ship. Co-Authored-By: Claude Opus 5.5 --- .../TransferBudgetReport.integration.ts | 51 ++++ .../TransferBudgetV2Fixture.integration.ts | 237 +++++++++++++++++- .../transferBudgetV2.integration.test.ts | 49 +++- docs/internals/performance-regressions.md | 18 +- 4 files changed, 343 insertions(+), 12 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 17988142dd9f..172bd9e6ebe5 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -11,11 +11,17 @@ import { TRANSFER_MEASURED_MCP_RESULT_BYTES, TRANSFER_MEASURED_TOOLS, } from "./fixtures/transferBudget.ts"; +import { LONG_TURN_WORK_ROWS } from "./TransferBudgetV2Fixture.integration.ts"; export interface WebSocketCatchUpMeasurement extends WebSocketTransferTotals { readonly mode: "replay" | "snapshot"; } +export interface TimelineHttpMeasurement { + readonly http: HttpTransferMeasurement; + readonly rows: number; +} + export interface TransferBudgetRun { readonly provider: ProviderDriverKind; readonly threadSnapshot: HttpTransferMeasurement; @@ -27,6 +33,9 @@ export interface TransferBudgetRun { readonly reconnectShell: WebSocketCatchUpMeasurement; readonly measuredTurnSqlStatements: number; readonly reconnectSqlStatements: number; + /** The compact bounded snapshot web and mobile load when the long-turn thread opens. */ + readonly longTurnColdOpen: TimelineHttpMeasurement; + readonly longTurnHistoryPage: TimelineHttpMeasurement; } interface ProviderTransferBudget { @@ -50,6 +59,15 @@ const TRANSFER_BUDGETS: Readonly> = { claudeAgent: TRANSFER_BUDGET, }; +// Decoded bytes, not wire bytes: phones parse and keep the decoded JSON. About +// 30% above the observed 92 KB; without row budgets inside user turns the cold +// open carried all 920 rows (1.07 MB). Kept apart from TRANSFER_BUDGET because +// the PR comment validates that object's keys. +const LONG_TURN_BUDGET = { + decodedBytes: 120_000, + rows: 75, +}; + function totalWireBytes(run: TransferBudgetRun): number { return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes; } @@ -144,6 +162,24 @@ function webSocketRows( ]; } +function longTurnRows( + provider: ProviderDriverKind, + phase: string, + measured: TimelineHttpMeasurement, +): string[] { + return [ + infoRow(provider, phase, "HTTP wire", measured.http.wireBytes), + row( + provider, + phase, + "HTTP decoded", + measured.http.decodedBodyBytes, + LONG_TURN_BUDGET.decodedBytes, + ), + row(provider, phase, "rows", measured.rows, LONG_TURN_BUDGET.rows, String), + ]; +} + export function transferBudgetViolations(runs: ReadonlyArray): string[] { const violations: string[] = []; for (const run of runs) { @@ -170,6 +206,18 @@ export function transferBudgetViolations(runs: ReadonlyArray) run.measuredTurnWebSocket.messages, budget.measuredTurnWebSocketMessages, ], + [ + "long-turn cold open decoded bytes", + run.longTurnColdOpen.http.decodedBodyBytes, + LONG_TURN_BUDGET.decodedBytes, + ], + ["long-turn cold open rows", run.longTurnColdOpen.rows, LONG_TURN_BUDGET.rows], + [ + "long-turn history page decoded bytes", + run.longTurnHistoryPage.http.decodedBodyBytes, + LONG_TURN_BUDGET.decodedBytes, + ], + ["long-turn history page rows", run.longTurnHistoryPage.rows, LONG_TURN_BUDGET.rows], ] as const; for (const [metric, observed, maximum] of checks) { if (observed > maximum) { @@ -187,6 +235,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray> { + const threadId = LONG_TURN_THREAD_ID; + const providerInstanceId = ProviderInstanceId.make(provider); + const modelSelection = { + instanceId: providerInstanceId, + model: provider === "codex" ? "gpt-5.4" : "claude-sonnet-4-6", + }; + let ordinal = 0; + const text = (turn: number, row: number, targetBytes: number) => + diagnosticOutput({ provider, turnIndex: 100 + turn, toolIndex: row, targetBytes }); + + const turn = (index: number, workRows: number): OrchestrationV2DomainEvent[] => { + const now = DateTime.makeUnsafe(`2026-06-02T00:${String(index).padStart(2, "0")}:00Z`); + const runId = RunId.make(`long-turn-run-${index}`); + const nodeId = NodeId.make(`long-turn-node-${index}`); + const common = { + threadId, + runId, + nodeId, + driver: provider, + providerInstanceId, + occurredAt: now, + }; + const run = { + id: runId, + threadId, + ordinal: index + 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`long-turn-user-${index}`), + rootNodeId: nodeId, + activeAttemptId: null, + status: "running" as const, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const base = (row: string, title: string | null) => { + ordinal += 1; + return { + id: TurnItemId.make(`long-turn-item-${index}-${row}`), + threadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed" as const, + title, + startedAt: now, + completedAt: now, + updatedAt: now, + }; + }; + const events: OrchestrationV2DomainEvent[] = [ + { ...common, id: EventId.make(`long-turn-run-${index}`), type: "run.created", payload: run }, + ]; + const message = (role: "user" | "assistant", row: string, messageText: string) => { + const messageId = MessageId.make(`long-turn-${role}-${index}-${row}`); + const createdBy = role === "user" ? ("user" as const) : ("agent" as const); + events.push( + { + ...common, + id: EventId.make(`long-turn-message-${index}-${row}`), + type: "message.updated", + payload: { + id: messageId, + threadId, + runId, + nodeId, + role, + text: messageText, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + createdBy, + creationSource: "web", + }, + }, + { + ...common, + id: EventId.make(`long-turn-item-event-${index}-${row}`), + type: "turn-item.updated", + payload: + role === "user" + ? { + ...base(row, null), + type: "user_message", + createdBy, + creationSource: "web", + messageId, + inputIntent: "turn_start", + text: messageText, + attachments: [], + } + : { + ...base(row, null), + type: "assistant_message", + messageId, + text: messageText, + streaming: false, + }, + }, + ); + }; + const workItem = (row: number): OrchestrationV2TurnItem => { + if (row % 100 === 50) { + return { + ...base(String(row), "Checkpoint"), + type: "checkpoint", + checkpointId: CheckpointId.make(`long-turn-checkpoint-${index}-${row}`), + scopeId: CheckpointScopeId.make(`long-turn-scope-${index}`), + files: Array.from({ length: 40 }, (_, file) => ({ + path: `src/feature/module-${row}-${file}.ts`, + kind: "modified", + additions: file + 1, + deletions: file, + })), + }; + } + if (row % 100 === 99) { + return { + ...base(String(row), "Subagent"), + type: "subagent", + subagentId: NodeId.make(`long-turn-subagent-${index}-${row}`), + origin: "app_owned", + driver: provider, + providerInstanceId, + childThreadId: null, + prompt: text(index, row, 2_048), + result: text(index, row + 1, 3_072), + }; + } + switch (row % 8) { + case 0: + case 7: + return { + ...base(String(row), "Reasoning"), + type: "reasoning", + text: text(index, row, 1_536), + streaming: false, + }; + case 3: + return { + ...base(String(row), `Edit module ${row}`), + type: "file_change", + fileName: `src/feature/module-${row}.ts`, + additions: 12, + deletions: 4, + diffStr: text(index, row, 2_048), + }; + case 4: + return { + ...base(String(row), "MCP search"), + type: "dynamic_tool", + toolName: "mcp__test__search", + input: { query: `module ${row}` }, + output: { text: text(index, row, 6_144) }, + }; + case 5: + return { + ...base(String(row), null), + type: "assistant_message", + messageId: MessageId.make(`long-turn-progress-${index}-${row}`), + text: text(index, row, 600), + streaming: false, + }; + default: + return { + ...base(String(row), `Run check ${row}`), + type: "command_execution", + input: `vp test run src/feature/module-${row}.test.ts`, + output: text(index, row, 4_096), + exitCode: 0, + }; + } + }; + + message("user", "prompt", `Continue the feature work, step ${index}.`); + for (let row = 0; row < workRows; row++) { + events.push({ + ...common, + id: EventId.make(`long-turn-work-${index}-${row}`), + type: "turn-item.updated", + payload: workItem(row), + }); + } + message("assistant", "answer", text(index, workRows, 1_200)); + events.push({ + ...common, + id: EventId.make(`long-turn-complete-${index}`), + type: "run.updated", + payload: { ...run, status: "completed", completedAt: now }, + }); + return events; + }; + + return [ + [threadCreated(provider, threadId)], + turn(0, 4), + turn(1, 4), + turn(2, 4), + turn(3, LONG_TURN_WORK_ROWS), + ]; +} diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 75b0be5ae57a..8417d852acfd 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -16,7 +16,9 @@ import { EnvironmentAuthenticatedPrincipal, ORCHESTRATION_V2_WS_METHODS, OrchestrationV2RpcSchemas, + OrchestrationV2ThreadBoundedSnapshot, OrchestrationV2ThreadDetailSnapshot, + OrchestrationV2ThreadHistoryPage, OrchestrationV2ShellSnapshot, OrchestrationV2GetThreadProjectionError, OrchestrationV2GetShellSnapshotError, @@ -63,11 +65,23 @@ import { type TransferBudgetRun, } from "./TransferBudgetReport.integration.ts"; import { TRANSFER_HISTORY_TURN_COUNT } from "./fixtures/transferBudget.ts"; -import { THREAD_ID, threadCreated, turnEvents } from "./TransferBudgetV2Fixture.integration.ts"; +import { + LONG_TURN_THREAD_ID, + longTurnThreadBatches, + THREAD_ID, + threadCreated, + turnEvents, +} from "./TransferBudgetV2Fixture.integration.ts"; const decodeThreadSnapshot = Schema.decodeUnknownEffect( Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot)), ); +const decodeBoundedSnapshot = Schema.decodeUnknownEffect( + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)), +); +const decodeHistoryPage = Schema.decodeUnknownEffect( + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage)), +); const decodeShellSnapshot = Schema.decodeUnknownEffect( Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellSnapshot)), ); @@ -87,6 +101,10 @@ const layerManagement = Layer.unwrap( getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), getThreadSnapshotWindow: (id, options) => projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), + getThreadHistoryPage: (id, cursor, throughEntryId, conversationOnly) => + projections + .getThreadHistoryPage(id, cursor, throughEntryId, conversationOnly) + .pipe(Effect.orDie), getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie), getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie), listActiveSubagentThreadIds: () => @@ -331,7 +349,29 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => const shellMode = yield* synchronized(rsq); const reconnectThread = { ...resumedThread.recorder.totals(), mode: threadMode }; const reconnectShell = { ...resumedShell.recorder.totals(), mode: shellMode }; + const reconnectSqlStatements = counter.count() - reconnectSqlStart; assert.equal(threadMode, "snapshot"); + + for (const events of longTurnThreadBatches(provider)) yield* sink.write({ events }); + const longTurnColdOpen = yield* measureHttpGet({ + url: `${base}/api/orchestration/threads/${LONG_TURN_THREAD_ID}/bounded?compactTurnItems=1`, + headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, + }); + assert.equal(longTurnColdOpen.status, 200); + const coldOpen = yield* decodeBoundedSnapshot( + Buffer.from(longTurnColdOpen.decodedBody).toString(), + ); + const cursor = coldOpen.historyCursor; + if (cursor === null) return yield* Effect.die("Expected older long-turn history"); + const longTurnHistoryPage = yield* measureHttpGet({ + url: `${base}/api/orchestration/threads/${LONG_TURN_THREAD_ID}/history?cursor=${encodeURIComponent(cursor)}`, + headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, + }); + assert.equal(longTurnHistoryPage.status, 200); + const historyPage = yield* decodeHistoryPage( + Buffer.from(longTurnHistoryPage.decodedBody).toString(), + ); + assert.isTrue(historyPage.hasMoreHistory); return { provider, threadSnapshot, @@ -342,7 +382,12 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => measuredTurnSqlStatements, reconnectThread, reconnectShell, - reconnectSqlStatements: counter.count() - reconnectSqlStart, + reconnectSqlStatements, + longTurnColdOpen: { + http: longTurnColdOpen, + rows: coldOpen.projection.visibleTurnItems.length, + }, + longTurnHistoryPage: { http: longTurnHistoryPage, rows: historyPage.items.length }, } satisfies TransferBudgetRun; }).pipe( Effect.provide(Layer.fresh(layerServices).pipe(Layer.provideMerge(NodeServices.layer))), diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 92ce89b7568f..1a9dfee13b82 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -12,8 +12,11 @@ client can produce reassuring timings while missing the active transport path. The v2 checks pin these invariants: -- Client cold opens target 75 recent timeline rows and about 1 MiB of encoded data. Required - control records and the at-least-one-row rule can exceed the byte target; it is not a hard cap. +- Cold opens and history pages stop at about 1 MiB of encoded rows. A window that holds a user + turn also stops at 10 user turns (20 on older pages) or 500 rows, so ordinary turns stay whole. + Any other window, including the inside of one turn longer than 500 rows, stops at 75 rows. The + snapshot query applies the same rule before it decodes payloads. Required control records and + the at-least-one-row rule can exceed the byte target; it is not a hard cap. - The bounded snapshot does not retain the full duplicate conversation-message table. - Older activity is fetched in bounded HTTP pages and merged without disturbing the live scroll window. @@ -29,8 +32,10 @@ The v2 checks pin these invariants: projects small can increase server allocations on repeated large tool updates. - Raw command output, dynamic-tool result bodies, and inline file-change bodies are omitted at the wire boundary, including small outputs. Explicit failure flags and bounded result IDs - preserve status and grouped action counts without shipping those bodies. Persisted events - remain complete; the existing diff endpoints still provide file content when requested. + preserve status and grouped action counts without shipping those bodies. Subagent prompts and + results, reasoning text, command lines, search results, and checkpoint file lists still ship. + Persisted events remain complete; the existing diff endpoints still provide file content when + requested. Explicit cross-environment transcript exports preserve full tool bodies so the destination agent can read the context. The export rejects timeline JSON above the file attachment limit before sending it; the client still applies its attachment limit to the complete file. @@ -57,7 +62,10 @@ transport performance fixtures validate contract encoding and compare pre-compre for the same synthetic workload, preserving historical application-object counts separately. These are neither compressed WebSocket captures nor measurements of server RSS, and command fixtures exclude unrelated settings calls and subscription events. Payload budgets belong in -these tests so regressions fail locally. +these tests so regressions fail locally. The transfer budget integration test +(`apps/server/integration/transferBudgetV2.integration.test.ts`) caps the decoded bytes and rows +of the compact bounded snapshot that web and mobile load, and of one older page, for a 900-row +agent turn. ## Event store and startup