From de22637d6c2f690402182ccad7331ce4041d142b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 26 Oct 2026 22:58:08 -0700 Subject: [PATCH 1/4] test(server): transfer budgets measure what current clients request The transfer budget test asked for thread and shell data the way older clients did, so the body dedupe, ID table, idle-subagent and older-settled savings went unmeasured. It now sends the current client opt-ins, uses realistic IDs and message records in the long-turn fixture, adds a busy sidebar fixture, and budgets the cold-connect shell. Co-Authored-By: Claude Opus 5.5 --- .../TransferBudgetReport.integration.ts | 72 +++++++++++++---- .../TransferBudgetV2Fixture.integration.ts | 77 ++++++++++++++---- .../transferBudgetV2.integration.test.ts | 81 ++++++++++++++----- docs/internals/performance-regressions.md | 7 +- 4 files changed, 181 insertions(+), 56 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index fbe8185a4b482..f6fb3fe671647 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 without the opt-ins current clients send, to show what they save. */ + 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 { @@ -59,15 +69,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 +175,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}, no opt-ins`, "HTTP wire", measured.http.wireBytes), + infoRow(provider, `${phase}, no opt-ins`, "HTTP decoded", measured.http.decodedBodyBytes), + infoRow(provider, `${phase}, no opt-ins`, "rows", measured.rows, String), ]; } @@ -216,6 +238,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 +262,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, }, }; } @@ -192,6 +197,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. + * 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 +215,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, @@ -237,14 +250,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, @@ -360,14 +374,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 +387,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 +416,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..294dd97d23b56 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -68,11 +68,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 +229,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)), @@ -353,25 +367,49 @@ 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 get = (path: string) => + measureHttpGet({ + url: `${base}${path}`, + headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, + }); + const longTurnPath = `/api/orchestration/threads/${LONG_TURN_THREAD_ID}`; + const coldOpen = yield* Effect.forEach( + ["compactTurnItems=1", CLIENT_BOUNDED_QUERY], + Effect.fnUntraced(function* (query) { + const http = yield* get(`${longTurnPath}/bounded?${query}`); + assert.equal(http.status, 200); + const snapshot = yield* decodeBoundedSnapshot( + Buffer.from(http.decodedBody).toString(), + ); + return { http, rows: snapshot.projection.visibleTurnItems.length, snapshot }; + }), ); - const cursor = coldOpen.historyCursor; + const cursor = coldOpen[1]!.snapshot.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(), + const historyPage = yield* Effect.forEach( + ["", `&${CLIENT_HISTORY_QUERY}`], + Effect.fnUntraced(function* (query) { + const http = yield* get( + `${longTurnPath}/history?cursor=${encodeURIComponent(cursor)}${query}`, + ); + assert.equal(http.status, 200); + const page = yield* decodeHistoryPage(Buffer.from(http.decodedBody).toString()); + assert.isTrue(page.hasMoreHistory); + return { http, rows: page.items.length }; + }), + ); + + yield* sink.write({ events: shellFleetEvents(provider) }); + const fleetShell = yield* Effect.forEach( + ["", `?${CLIENT_SHELL_QUERY}`], + Effect.fnUntraced(function* (query) { + const http = yield* get(`/api/orchestration/shell${query}`); + assert.equal(http.status, 200); + const snapshot = yield* decodeShellSnapshot(Buffer.from(http.decodedBody).toString()); + return { http, rows: snapshot.threads.length }; + }), ); - assert.isTrue(historyPage.hasMoreHistory); + assert.equal(fleetShell[0]!.rows, 2 + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS); return { provider, threadSnapshot, @@ -383,11 +421,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: coldOpen[0]!, + longTurnColdOpen: coldOpen[1]!, + longTurnHistoryPageLegacy: historyPage[0]!, + longTurnHistoryPage: historyPage[1]!, + fleetShellLegacy: fleetShell[0]!, + fleetShell: fleetShell[1]!, } 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..1ed06186cbe4a 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, or its savings go unmeasured. ## Event store and startup From a103e1f29fec732d1d892c12978f8639ba8c2fe6 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 26 Oct 2026 23:21:54 -0700 Subject: [PATCH 2/4] test(server): budget live body dedupe and name the transfer reads The measured live turn now repeats its answer in a turn item, as adapters do, so the live deduper has work and its loss fails the decoded budget. Reads are named instead of positional, the comparison rows say "older client", and message IDs use the adapter's composite shape. Co-Authored-By: Claude Opus 5.5 --- .../TransferBudgetReport.integration.ts | 11 +-- .../TransferBudgetV2Fixture.integration.ts | 64 ++++++++++------ .../transferBudgetV2.integration.test.ts | 73 ++++++++++--------- docs/internals/performance-regressions.md | 2 +- 4 files changed, 87 insertions(+), 63 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index f6fb3fe671647..7537ace226766 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -56,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; @@ -194,9 +195,9 @@ function legacyRows( measured: TimelineHttpMeasurement, ): string[] { return [ - infoRow(provider, `${phase}, no opt-ins`, "HTTP wire", measured.http.wireBytes), - infoRow(provider, `${phase}, no opt-ins`, "HTTP decoded", measured.http.decodedBodyBytes), - infoRow(provider, `${phase}, no opt-ins`, "rows", measured.rows, String), + 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), ]; } @@ -262,7 +263,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray ({ + 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({ @@ -145,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}`), @@ -182,6 +186,20 @@ export function turnEvents( }, }); } + // The live turn repeats its answer in a turn item, as adapters do, so live dedupe has work. + 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}`), @@ -272,7 +290,9 @@ 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 messageId = MessageId.make( + `message:provider:${provider}:native-item:msg_01LongTurn${index}x${row}`, + ); const createdBy = role === "user" ? ("user" as const) : ("agent" as const); events.push( { diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 294dd97d23b56..b08a08fa9d094 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"; @@ -371,45 +372,47 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => 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 coldOpen = yield* Effect.forEach( - ["compactTurnItems=1", CLIENT_BOUNDED_QUERY], - Effect.fnUntraced(function* (query) { + const readColdOpen = (query: string) => + Effect.gen(function* () { const http = yield* get(`${longTurnPath}/bounded?${query}`); - assert.equal(http.status, 200); - const snapshot = yield* decodeBoundedSnapshot( - Buffer.from(http.decodedBody).toString(), - ); - return { http, rows: snapshot.projection.visibleTurnItems.length, snapshot }; - }), - ); - const cursor = coldOpen[1]!.snapshot.historyCursor; - if (cursor === null) return yield* Effect.die("Expected older long-turn history"); - const historyPage = yield* Effect.forEach( - ["", `&${CLIENT_HISTORY_QUERY}`], - Effect.fnUntraced(function* (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}`, ); - assert.equal(http.status, 200); - const page = yield* decodeHistoryPage(Buffer.from(http.decodedBody).toString()); + 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 longTurnColdOpenLegacy = yield* readColdOpen("compactTurnItems=1"); + const longTurnColdOpen = yield* readColdOpen(CLIENT_BOUNDED_QUERY); + const { cursor } = longTurnColdOpen; + if (cursor === null) return yield* Effect.die("Expected older long-turn history"); + const longTurnHistoryPageLegacy = yield* readHistoryPage(cursor, ""); + const longTurnHistoryPage = yield* readHistoryPage(cursor, `&${CLIENT_HISTORY_QUERY}`); + const earlierThreads = decodedShell.threads.length + 1; yield* sink.write({ events: shellFleetEvents(provider) }); - const fleetShell = yield* Effect.forEach( - ["", `?${CLIENT_SHELL_QUERY}`], - Effect.fnUntraced(function* (query) { - const http = yield* get(`/api/orchestration/shell${query}`); - assert.equal(http.status, 200); - const snapshot = yield* decodeShellSnapshot(Buffer.from(http.decodedBody).toString()); - return { http, rows: snapshot.threads.length }; - }), + const fleetShellLegacy = yield* readShell(""); + const fleetShell = yield* readShell(`?${CLIENT_SHELL_QUERY}`); + assert.equal( + fleetShellLegacy.rows, + earlierThreads + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS, ); - assert.equal(fleetShell[0]!.rows, 2 + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS); return { provider, threadSnapshot, @@ -421,12 +424,12 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => reconnectThread, reconnectShell, reconnectSqlStatements, - longTurnColdOpenLegacy: coldOpen[0]!, - longTurnColdOpen: coldOpen[1]!, - longTurnHistoryPageLegacy: historyPage[0]!, - longTurnHistoryPage: historyPage[1]!, - fleetShellLegacy: fleetShell[0]!, - fleetShell: fleetShell[1]!, + 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 1ed06186cbe4a..51760756c1085 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -66,7 +66,7 @@ these tests so regressions fail locally. The transfer budget integration test (`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, or its savings go unmeasured. +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 From 56c540752dd1824baef30f66a225e7fcb5afe84f Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 26 Oct 2026 23:26:45 -0700 Subject: [PATCH 3/4] test(server): assert the live answer arrives deduped The byte budget alone passed live-dedupe loss by about 1 KB. The test now checks that the repeated answer ships blank and marked. Read helpers take a bare query, and the fixture comments say why only the live turn repeats its answer. Co-Authored-By: Claude Opus 5.5 --- .../TransferBudgetV2Fixture.integration.ts | 5 +++-- .../transferBudgetV2.integration.test.ts | 22 ++++++++++++++----- 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/apps/server/integration/TransferBudgetV2Fixture.integration.ts b/apps/server/integration/TransferBudgetV2Fixture.integration.ts index 72a1e47287848..c4bac66a8959a 100644 --- a/apps/server/integration/TransferBudgetV2Fixture.integration.ts +++ b/apps/server/integration/TransferBudgetV2Fixture.integration.ts @@ -187,6 +187,7 @@ 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, @@ -215,8 +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. - * IDs follow the composite shapes the Claude adapter stores, and every - * assistant message has its record, as real threads do. + * 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, diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index b08a08fa9d094..0ab61b62cd5c1 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -338,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); @@ -386,7 +397,7 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => const readHistoryPage = (cursor: string, query: string) => Effect.gen(function* () { const http = yield* get( - `${longTurnPath}/history?cursor=${encodeURIComponent(cursor)}${query}`, + `${longTurnPath}/history?cursor=${encodeURIComponent(cursor)}&${query}`, ); const page = yield* decodeHistoryPage(decoded(http)); assert.isTrue(page.hasMoreHistory); @@ -394,7 +405,7 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => }); const readShell = (query: string) => Effect.gen(function* () { - const http = yield* get(`/api/orchestration/shell${query}`); + const http = yield* get(`/api/orchestration/shell?${query}`); return { http, rows: (yield* decodeShellSnapshot(decoded(http))).threads.length }; }); @@ -403,12 +414,13 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => const { cursor } = longTurnColdOpen; if (cursor === null) return yield* Effect.die("Expected older long-turn history"); const longTurnHistoryPageLegacy = yield* readHistoryPage(cursor, ""); - const longTurnHistoryPage = yield* readHistoryPage(cursor, `&${CLIENT_HISTORY_QUERY}`); + 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}`); + const fleetShell = yield* readShell(CLIENT_SHELL_QUERY); assert.equal( fleetShellLegacy.rows, earlierThreads + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS, From fd02199cad33d154d9a159c5892a0c09b435af5f Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 26 Oct 2026 23:29:09 -0700 Subject: [PATCH 4/4] test(server): tidy transfer budget reads and long-turn message IDs Co-Authored-By: Claude Opus 5.5 --- .../TransferBudgetReport.integration.ts | 2 +- .../TransferBudgetV2Fixture.integration.ts | 16 +++++++++------- .../transferBudgetV2.integration.test.ts | 6 +++--- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 7537ace226766..fd434e2bc26eb 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -40,7 +40,7 @@ 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 without the opt-ins current clients send, to show what they save. */ + /** 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. */ diff --git a/apps/server/integration/TransferBudgetV2Fixture.integration.ts b/apps/server/integration/TransferBudgetV2Fixture.integration.ts index c4bac66a8959a..8fab29ad64d38 100644 --- a/apps/server/integration/TransferBudgetV2Fixture.integration.ts +++ b/apps/server/integration/TransferBudgetV2Fixture.integration.ts @@ -250,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, @@ -257,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, @@ -291,9 +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( - `message:provider:${provider}:native-item:msg_01LongTurn${index}x${row}`, - ); + const id = messageId(role, row); const createdBy = role === "user" ? ("user" as const) : ("agent" as const); events.push( { @@ -301,7 +303,7 @@ export function longTurnThreadBatches( id: EventId.make(`long-turn-message-${index}-${row}`), type: "message.updated", payload: { - id: messageId, + id, threadId, runId, nodeId, @@ -326,7 +328,7 @@ export function longTurnThreadBatches( type: "user_message", createdBy, creationSource: "web", - messageId, + messageId: id, inputIntent: "turn_start", text: messageText, attachments: [], @@ -334,7 +336,7 @@ export function longTurnThreadBatches( : { ...base(row, null), type: "assistant_message", - messageId, + messageId: id, text: messageText, streaming: false, }, diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 0ab61b62cd5c1..22db0e643db9b 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -409,9 +409,9 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => return { http, rows: (yield* decodeShellSnapshot(decoded(http))).threads.length }; }); - const longTurnColdOpenLegacy = yield* readColdOpen("compactTurnItems=1"); - const longTurnColdOpen = yield* readColdOpen(CLIENT_BOUNDED_QUERY); - const { cursor } = longTurnColdOpen; + 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 longTurnHistoryPageLegacy = yield* readHistoryPage(cursor, ""); const longTurnHistoryPage = yield* readHistoryPage(cursor, CLIENT_HISTORY_QUERY);