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/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 14fed37c0618..419e4f9d4eb2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -40,8 +40,10 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import { buildBoundedThreadProjection, decodeThreadHistoryCursor, + encodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY, + THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, } from "./threadHistoryPaging.ts"; const layerTest = Layer.mergeAll( @@ -1529,6 +1531,126 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { [baselineId], ); } + 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` + 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 + `; + + // 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 ( @@ -1571,9 +1693,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 +1708,14 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const anchorItemId = TurnItemId.make(decodeThreadHistoryCursor(cursor).si); const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, + userTurnLimit: 20, anchorItemId, }); + assert.isAtMost( + snapshot.projection.visibleTurnItems.length, + THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, + ); const page = selectHistoryPageFromCursor({ - policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined }, items: snapshot.projection.visibleTurnItems, cursor, snapshotSequence: snapshot.snapshotSequence, @@ -1599,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 438cf79138dd..4ddc34ae4a4f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -96,10 +96,12 @@ import { isConversationHistoryItem, threadHistoryTurnStarts, THREAD_HISTORY_MAX_RAW_TURNS, + missingInterruptRequests, + olderHistoryPagePolicy, 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()( @@ -2992,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 +3015,9 @@ export const layer: Layer.Layer = 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 + 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 ( @@ -5478,84 +5493,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), - }; - }), - ); - if (hasUserTurns) - return { - ...page, - snapshotSequence, - items: items.map((row, position) => ({ ...row, position })), - }; - const bounded = selectRecentTimelineWindow({ - items, - snapshotSequence, - policy: { - ...THREAD_HISTORY_PAGE_POLICY, - maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, - maxItems: - THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2), - }, + const 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, }; @@ -7383,21 +7399,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 }), @@ -7448,7 +7469,16 @@ export const layerMemory: Layer.Layer = Layer.effect( ? rawStart : anchors.at(-anchorLimit)! : Math.max(0, anchorIndex - options.rowLimit); - const visibleTurnItems = candidates.slice(start); + 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, rowLimit)), + ); return { ...snapshot, projection: { ...snapshot.projection, visibleTurnItems }, 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; } 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 4055fd85b7ae..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,6 +155,65 @@ function makeProjection(visibleTurnItems: OrchestrationV2ProjectedTurnItem[]) { } describe("threadHistoryPaging", () => { + 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]; + 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( + page.items[0]?.item.type === "user_message" + ? THREAD_HISTORY_PAGE_POLICY.maxTurnItems + : 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(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 +233,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 +284,20 @@ 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, + 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}`); 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 +335,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", () => { @@ -557,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); @@ -566,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, @@ -585,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); @@ -641,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", () => { @@ -685,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 e0aa215f3f1e..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"; /** - * Item/byte budgets only apply to histories without turn starts; tool activity - * must not split an ordinary conversation turn. + * 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); } @@ -210,8 +195,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; @@ -236,14 +220,17 @@ 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 = 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 >= maxItems || + encodedBytes + rowBytes > policy.maxEncodedBytes || + (turnLimit !== undefined && + (userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS))) ) { break; } @@ -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]; @@ -307,6 +298,44 @@ 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, + maxTurnItems: (policy.maxTurnItems ?? policy.maxItems) * 2, + }; +} + export function selectHistoryPageFromCursor(input: { readonly items: ReadonlyArray; readonly cursor: string; @@ -316,41 +345,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, @@ -529,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, @@ -604,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/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 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/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 a772eaaf76c2..9956694aa193 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({ @@ -1084,7 +1149,6 @@ describe("EnvironmentThreads", () => { historyCursor: "socket-history-cursor", hasMoreHistory: true, latestLocalTurnOrdinal: 47, - payloadBudgetExceeded: false, }); const state = yield* awaitThreadState( @@ -1199,7 +1263,6 @@ describe("EnvironmentThreads", () => { historyCursor: "progressive-cursor", hasMoreHistory: true, latestLocalTurnOrdinal: 47, - payloadBudgetExceeded: false, }); yield* awaitThreadState( harness.observed, @@ -1308,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({