diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index fbe8185a4b482..fd434e2bc26eb 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -11,7 +11,11 @@ import { TRANSFER_MEASURED_MCP_RESULT_BYTES, TRANSFER_MEASURED_TOOLS, } from "./fixtures/transferBudget.ts"; -import { LONG_TURN_WORK_ROWS } from "./TransferBudgetV2Fixture.integration.ts"; +import { + LONG_TURN_WORK_ROWS, + SHELL_IDLE_SUBAGENTS, + SHELL_SETTLED_THREADS, +} from "./TransferBudgetV2Fixture.integration.ts"; export interface WebSocketCatchUpMeasurement extends WebSocketTransferTotals { readonly mode: "replay" | "snapshot"; @@ -36,6 +40,12 @@ export interface TransferBudgetRun { /** The compact bounded snapshot web and mobile load when the long-turn thread opens. */ readonly longTurnColdOpen: TimelineHttpMeasurement; readonly longTurnHistoryPage: TimelineHttpMeasurement; + /** The same reads as an older client that only sends `compactTurnItems`, to show the savings. */ + readonly longTurnColdOpenLegacy: TimelineHttpMeasurement; + readonly longTurnHistoryPageLegacy: TimelineHttpMeasurement; + /** The cold-connect shell snapshot after the settled-thread fleet is written. */ + readonly fleetShell: TimelineHttpMeasurement; + readonly fleetShellLegacy: TimelineHttpMeasurement; } interface ProviderTransferBudget { @@ -46,11 +56,12 @@ interface ProviderTransferBudget { readonly measuredTurnWebSocketMessages: number; } +// The measured turn decodes to about 22 KB; without live body dedupe its repeated answer makes it 26 KB. const TRANSFER_BUDGET = { totalWireBytes: 7_000, threadSnapshotWireBytes: 5_000, measuredTurnWebSocketWireBytes: 2_000, - measuredTurnWebSocketDecodedBytes: 30_000, + measuredTurnWebSocketDecodedBytes: 25_000, measuredTurnWebSocketMessages: 8, } satisfies ProviderTransferBudget; @@ -59,15 +70,20 @@ 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. +// Decoded bytes, not wire bytes: phones parse and keep the decoded JSON. About 7% above the +// observed 103 KB, so a cold open without body dedupe (117 KB), the ID table, or row budgets +// (all 920 rows) fails. Kept apart from TRANSFER_BUDGET because the PR comment validates its keys. const LONG_TURN_BUDGET = { - decodedBytes: 120_000, + decodedBytes: 110_000, rows: 75, }; +// About 30% above the observed 42 KB. Without the shell opt-ins it carries all 142 rows (217 KB). +const FLEET_SHELL_BUDGET = { + decodedBytes: 55_000, + rows: 27, +}; + function totalWireBytes(run: TransferBudgetRun): number { return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes; } @@ -160,21 +176,28 @@ function webSocketRows( ]; } -function longTurnRows( +function timelineRows( provider: ProviderDriverKind, phase: string, measured: TimelineHttpMeasurement, + budget: { readonly decodedBytes: number; readonly rows: number }, ): 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), + row(provider, phase, "HTTP decoded", measured.http.decodedBodyBytes, budget.decodedBytes), + row(provider, phase, "rows", measured.rows, budget.rows, String), + ]; +} + +function legacyRows( + provider: ProviderDriverKind, + phase: string, + measured: TimelineHttpMeasurement, +): string[] { + return [ + infoRow(provider, `${phase}, older client`, "HTTP wire", measured.http.wireBytes), + infoRow(provider, `${phase}, older client`, "HTTP decoded", measured.http.decodedBodyBytes), + infoRow(provider, `${phase}, older client`, "rows", measured.rows, String), ]; } @@ -216,6 +239,12 @@ export function transferBudgetViolations(runs: ReadonlyArray) LONG_TURN_BUDGET.decodedBytes, ], ["long-turn history page rows", run.longTurnHistoryPage.rows, LONG_TURN_BUDGET.rows], + [ + "fleet shell snapshot decoded bytes", + run.fleetShell.http.decodedBodyBytes, + FLEET_SHELL_BUDGET.decodedBytes, + ], + ["fleet shell snapshot rows", run.fleetShell.rows, FLEET_SHELL_BUDGET.rows], ] as const; for (const [metric, observed, maximum] of checks) { if (observed > maximum) { @@ -234,6 +263,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray; export function threadCreated( provider: ProviderDriverKind, threadId: ThreadId = THREAD_ID, -): OrchestrationV2DomainEvent { + overrides: Partial = {}, +): ThreadCreatedEvent { const now = DateTime.makeUnsafe("2026-06-01T00:00:00Z"); return { id: EventId.make(`thread-created:${threadId}`), @@ -58,6 +62,7 @@ export function threadCreated( settledAt: null, lastVisitedAt: null, deletedAt: null, + ...overrides, }, }; } @@ -102,6 +107,9 @@ export function turnEvents( const events: OrchestrationV2DomainEvent[] = [ { ...common, id: EventId.make(`run-${index}`), type: "run.created", payload: run }, ]; + const answer = + `Completed ${provider} inspection ${index}. ` + + "The projection preserves transcript text and bounds tool output. ".repeat(60); for (const role of ["user", "assistant"] as const) { events.push({ ...common, @@ -113,11 +121,7 @@ export function turnEvents( runId, nodeId, role, - text: - role === "user" - ? `Inspect transfer ${index}` - : `Completed ${provider} inspection ${index}. ` + - "The projection preserves transcript text and bounds tool output. ".repeat(60), + text: role === "user" ? `Inspect transfer ${index}` : answer, attachments: [], streaming: false, createdAt: now, @@ -128,6 +132,22 @@ export function turnEvents( }); } const toolCount = measured ? TRANSFER_MEASURED_TOOLS : TRANSFER_HISTORY_TOOLS_PER_TURN; + const item = (id: string, ordinal: number, title: string | null) => ({ + id: TurnItemId.make(id), + threadId: THREAD_ID, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed" as const, + title, + startedAt: now, + completedAt: now, + updatedAt: now, + }); for (let tool = 0; tool <= toolCount; tool++) { const mcp = tool === toolCount; const output = diagnosticOutput({ @@ -140,22 +160,11 @@ export function turnEvents( : TRANSFER_HISTORY_MCP_RESULT_BYTES : 4096, }); - const base = { - id: TurnItemId.make(`item-${index}-${tool}`), - threadId: THREAD_ID, - runId, - nodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: tool + 1, - status: "completed" as const, - title: mcp ? "MCP search" : `Inspect source ${tool}`, - startedAt: now, - completedAt: now, - updatedAt: now, - }; + const base = item( + `item-${index}-${tool}`, + tool + 1, + mcp ? "MCP search" : `Inspect source ${tool}`, + ); events.push({ ...common, id: EventId.make(`tool-${index}-${tool}`), @@ -177,6 +186,21 @@ export function turnEvents( }, }); } + // The live turn repeats its answer in a turn item, as adapters do, so live dedupe has work. + // History turns skip it to keep the fallback thread snapshot under its wire ceiling. + if (measured) + events.push({ + ...common, + id: EventId.make(`answer-item-${index}`), + type: "turn-item.updated", + payload: { + ...item(`item-${index}-answer`, toolCount + 2, null), + type: "assistant_message", + messageId: MessageId.make(`assistant-${index}`), + text: answer, + streaming: false, + }, + }); events.push({ ...common, id: EventId.make(`complete-${index}`), @@ -192,6 +216,8 @@ export const LONG_TURN_WORK_ROWS = 900; /** * Three short user turns, then one agent turn far longer than a cold-open * window, with the item kinds a long agent run produces. One batch per turn. + * Run, node, provider and item IDs follow the composite shapes the Claude + * adapter stores, and every assistant message has its record, as real threads do. */ export function longTurnThreadBatches( provider: ProviderDriverKind, @@ -208,8 +234,14 @@ export function longTurnThreadBatches( 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 runId = RunId.make(`run:thread:${threadId}:ordinal:${index + 1}`); + const nodeId = NodeId.make(`node:run:${encodeURIComponent(runId)}:root`); + const providerThreadId = ProviderThreadId.make( + `provider-thread:provider:${provider}:native-thread:${encodeURIComponent(`pending:${runId}`)}`, + ); + const providerTurnId = ProviderTurnId.make( + `provider-turn:provider:${provider}:native-turn:${encodeURIComponent(`turn:run-attempt:${encodeURIComponent(runId)}:attempt:1`)}`, + ); const common = { threadId, runId, @@ -218,6 +250,10 @@ export function longTurnThreadBatches( providerInstanceId, occurredAt: now, }; + const messageId = (role: "user" | "assistant", row: string) => + MessageId.make( + `message:provider:${provider}:native-item:msg_01LongTurn${role}${index}x${row}`, + ); const run = { id: runId, threadId, @@ -225,7 +261,7 @@ export function longTurnThreadBatches( providerInstanceId, modelSelection, providerThreadId: null, - userMessageId: MessageId.make(`long-turn-user-${index}`), + userMessageId: messageId("user", "prompt"), rootNodeId: nodeId, activeAttemptId: null, status: "running" as const, @@ -237,14 +273,15 @@ export function longTurnThreadBatches( }; const base = (row: string, title: string | null) => { ordinal += 1; + const nativeId = `toolu_01LongTurn${index}x${row}`; return { - id: TurnItemId.make(`long-turn-item-${index}-${row}`), + id: TurnItemId.make(`turn-item:provider:${provider}:native-item:${nativeId}`), threadId, runId, nodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, + providerThreadId, + providerTurnId, + nativeItemRef: { driver: provider, nativeId, strength: "strong" as const }, parentItemId: null, ordinal, status: "completed" as const, @@ -258,7 +295,7 @@ export function longTurnThreadBatches( { ...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 id = messageId(role, row); const createdBy = role === "user" ? ("user" as const) : ("agent" as const); events.push( { @@ -266,7 +303,7 @@ export function longTurnThreadBatches( id: EventId.make(`long-turn-message-${index}-${row}`), type: "message.updated", payload: { - id: messageId, + id, threadId, runId, nodeId, @@ -291,7 +328,7 @@ export function longTurnThreadBatches( type: "user_message", createdBy, creationSource: "web", - messageId, + messageId: id, inputIntent: "turn_start", text: messageText, attachments: [], @@ -299,7 +336,7 @@ export function longTurnThreadBatches( : { ...base(row, null), type: "assistant_message", - messageId, + messageId: id, text: messageText, streaming: false, }, @@ -360,14 +397,6 @@ export function longTurnThreadBatches( 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}`), @@ -381,6 +410,10 @@ export function longTurnThreadBatches( message("user", "prompt", `Continue the feature work, step ${index}.`); for (let row = 0; row < workRows; row++) { + if (row % 8 === 5) { + message("assistant", String(row), text(index, row, 1_500)); + continue; + } events.push({ ...common, id: EventId.make(`long-turn-work-${index}-${row}`), @@ -406,3 +439,40 @@ export function longTurnThreadBatches( turn(3, LONG_TURN_WORK_ROWS), ]; } + +export const SHELL_SETTLED_THREADS = 100; +export const SHELL_IDLE_SUBAGENTS = 40; + +/** + * A busy sidebar: settled top-level threads, settled a minute apart, with idle subagent children + * under the newest ones. Opted-in cold connects skip the children and all but 25 settled threads. + */ +export function shellFleetEvents( + provider: ProviderDriverKind, +): ReadonlyArray { + return Array.from({ length: SHELL_SETTLED_THREADS }, (_, index) => { + const threadId = ThreadId.make(`transfer-budget-settled-${index}`); + const settledAt = DateTime.makeUnsafe(Date.UTC(2026, 5, 3, 0, index)); + const settled = threadCreated(provider, threadId, { + title: `Settled transfer thread ${index}`, + branch: `feature/settled-${index}`, + settledOverride: "settled", + settledAt, + updatedAt: settledAt, + }); + if (index < SHELL_SETTLED_THREADS - SHELL_IDLE_SUBAGENTS) return [settled]; + const childId = ThreadId.make(`transfer-budget-settled-${index}-subagent`); + return [ + settled, + threadCreated(provider, childId, { + title: `Subagent of settled thread ${index}`, + createdBy: "agent", + lineage: { + parentThreadId: threadId, + relationshipToParent: "subagent", + rootThreadId: threadId, + }, + }), + ]; + }).flat(); +} diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 8417d852acfdf..22db0e643db9b 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -55,6 +55,7 @@ import { subscribeOrchestrationV2Thread, subscribeOrchestrationV2Shell } from ". import { measureHttpGet, openMeasuredWsClient, + type HttpTransferMeasurement, type MeasuredWsClient, } from "./NetworkTransferMeasurement.integration.ts"; import { makeSqlStatementCounter } from "./SqlStatementCounter.integration.ts"; @@ -68,11 +69,20 @@ import { TRANSFER_HISTORY_TURN_COUNT } from "./fixtures/transferBudget.ts"; import { LONG_TURN_THREAD_ID, longTurnThreadBatches, + SHELL_IDLE_SUBAGENTS, + SHELL_SETTLED_THREADS, + shellFleetEvents, THREAD_ID, threadCreated, turnEvents, } from "./TransferBudgetV2Fixture.integration.ts"; +// The opt-ins current web and mobile clients send; keep in sync with client-runtime's +// boundedThreadSnapshotHttp.ts, threadHistoryHttp.ts, shellSnapshotHttp.ts, threads.ts and shell.ts. +const CLIENT_BOUNDED_QUERY = "compactTurnItems=1&dedupeBodies=1&compactIds=1"; +const CLIENT_HISTORY_QUERY = "compactIds=1"; +const CLIENT_SHELL_QUERY = "omitIdleSubagents=1&omitOlderSettled=1"; + const decodeThreadSnapshot = Schema.decodeUnknownEffect( Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot)), ); @@ -220,10 +230,15 @@ const subscribe = Effect.fn("transfer.subscribe")(function* ( afterSequence, requestCompletionMarker: true, acceptBoundedSnapshot: true, + acceptCompactTurnItems: true, + acceptCompactIds: true, + acceptDedupedBodies: true, }) : client.client[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({ afterSequence, requestCompletionMarker: true, + omitIdleSubagents: true, + omitOlderSettled: true, }); yield* stream.pipe( Stream.runForEach((item) => Queue.offer(queue, item)), @@ -323,7 +338,18 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => const finalShell = yield* projection.getThreadShell(THREAD_ID); const shellTerminal = (item: StreamItem) => item.kind === "thread.updated" && NodeUtil.isDeepStrictEqual(item.thread, finalShell); - for (const queue of [tq, ttq]) yield* collectUntil(queue, threadTerminal); + const liveTurn = yield* collectUntil(tq, threadTerminal); + yield* collectUntil(ttq, threadTerminal); + // A direct check, so live dedupe stays guarded however small the turn gets. + const liveAnswer = liveTurn.flatMap((item) => + item.kind === "event" && + item.event.type === "turn-item.updated" && + item.event.payload.type === "assistant_message" + ? [item.event.payload] + : [], + ); + assert.equal(liveAnswer.length, 1); + assert.equal(liveAnswer[0]!.textOmitted, true); for (const queue of [sq, ssq]) yield* collectUntil(queue, shellTerminal); const measuredTurnWebSocket = difference(thread.recorder.totals(), threadBefore); const measuredTurnShellWebSocket = difference(shell.recorder.totals(), shellBefore); @@ -353,25 +379,52 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => 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; + const get = (path: string) => + measureHttpGet({ + url: `${base}${path}`, + headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, + }).pipe(Effect.tap((http) => Effect.sync(() => assert.equal(http.status, 200)))); + const decoded = (http: HttpTransferMeasurement) => + Buffer.from(http.decodedBody).toString(); + const longTurnPath = `/api/orchestration/threads/${LONG_TURN_THREAD_ID}`; + const readColdOpen = (query: string) => + Effect.gen(function* () { + const http = yield* get(`${longTurnPath}/bounded?${query}`); + const snapshot = yield* decodeBoundedSnapshot(decoded(http)); + const rows = snapshot.projection.visibleTurnItems.length; + return { http, rows, cursor: snapshot.historyCursor }; + }); + const readHistoryPage = (cursor: string, query: string) => + Effect.gen(function* () { + const http = yield* get( + `${longTurnPath}/history?cursor=${encodeURIComponent(cursor)}&${query}`, + ); + const page = yield* decodeHistoryPage(decoded(http)); + assert.isTrue(page.hasMoreHistory); + return { http, rows: page.items.length }; + }); + const readShell = (query: string) => + Effect.gen(function* () { + const http = yield* get(`/api/orchestration/shell?${query}`); + return { http, rows: (yield* decodeShellSnapshot(decoded(http))).threads.length }; + }); + + const { cursor: _, ...longTurnColdOpenLegacy } = + yield* readColdOpen("compactTurnItems=1"); + const { cursor, ...longTurnColdOpen } = yield* readColdOpen(CLIENT_BOUNDED_QUERY); 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(), + const longTurnHistoryPageLegacy = yield* readHistoryPage(cursor, ""); + const longTurnHistoryPage = yield* readHistoryPage(cursor, CLIENT_HISTORY_QUERY); + + // The transfer thread plus the long-turn thread. + const earlierThreads = decodedShell.threads.length + 1; + yield* sink.write({ events: shellFleetEvents(provider) }); + const fleetShellLegacy = yield* readShell(""); + const fleetShell = yield* readShell(CLIENT_SHELL_QUERY); + assert.equal( + fleetShellLegacy.rows, + earlierThreads + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS, ); - assert.isTrue(historyPage.hasMoreHistory); return { provider, threadSnapshot, @@ -383,11 +436,12 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => reconnectThread, reconnectShell, reconnectSqlStatements, - longTurnColdOpen: { - http: longTurnColdOpen, - rows: coldOpen.projection.visibleTurnItems.length, - }, - longTurnHistoryPage: { http: longTurnHistoryPage, rows: historyPage.items.length }, + longTurnColdOpenLegacy, + longTurnColdOpen, + longTurnHistoryPageLegacy, + longTurnHistoryPage, + fleetShellLegacy, + fleetShell, } 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 1a9dfee13b82b..51760756c1085 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -63,9 +63,10 @@ for the same synthetic workload, preserving historical application-object counts 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. 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. +(`apps/server/integration/transferBudgetV2.integration.test.ts`) sends the opt-ins current clients +send and caps the decoded bytes and rows of the bounded snapshot and one older page for a 900-row +agent turn, and of the cold-connect shell for a sidebar of settled threads and idle subagents. When +clients gain an opt-in, add it to the test's client queries and give a fixture the data it saves. ## Event store and startup