From 9b21b26d37bf4f555ab806d873ff820e78d48504 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 11:11:05 -0700 Subject: [PATCH 01/10] test(server): guard basic-use transfer budgets --- .github/workflows/ci.yml | 11 + .../NetworkTransferMeasurement.integration.ts | 177 +++++++++++ .../TestProviderAdapter.integration.ts | 23 +- .../TransferBudgetReport.integration.ts | 206 +++++++++++++ .../TransferBudgetScenario.integration.ts | 130 +++++++++ .../integration/fixtures/transferBudget.ts | 202 +++++++++++++ apps/server/src/server.test.ts | 276 ++++++++++++++++-- 7 files changed, 985 insertions(+), 40 deletions(-) create mode 100644 apps/server/integration/NetworkTransferMeasurement.integration.ts create mode 100644 apps/server/integration/TransferBudgetReport.integration.ts create mode 100644 apps/server/integration/TransferBudgetScenario.integration.ts create mode 100644 apps/server/integration/fixtures/transferBudget.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1e51867cbe7..4845f63a1e8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -84,8 +84,19 @@ jobs: run: vp run --filter @t3tools/desktop ensure:electron - name: Test + env: + T3CODE_TRANSFER_BUDGET_REPORT_PATH: ${{ runner.temp }}/t3code-transfer-budget.md run: vp run test + - name: Publish transfer budget report + if: always() + run: | + if test -f "${{ runner.temp }}/t3code-transfer-budget.md"; then + tee -a "$GITHUB_STEP_SUMMARY" < "${{ runner.temp }}/t3code-transfer-budget.md" + else + echo "Transfer budget report was not produced." >> "$GITHUB_STEP_SUMMARY" + fi + - name: Test resource monitor run: cargo test --locked --manifest-path native/resource-monitor/Cargo.toml diff --git a/apps/server/integration/NetworkTransferMeasurement.integration.ts b/apps/server/integration/NetworkTransferMeasurement.integration.ts new file mode 100644 index 00000000000..75714d1519e --- /dev/null +++ b/apps/server/integration/NetworkTransferMeasurement.integration.ts @@ -0,0 +1,177 @@ +// @effect-diagnostics nodeBuiltinImport:off - Measures the real Node HTTP and WebSocket transports. +import * as NodeHttp from "node:http"; +import * as NodeZlib from "node:zlib"; + +import * as NodeSocket from "@effect/platform-node/NodeSocket"; +import { WsRpcGroup } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import { RpcClient, RpcSerialization } from "effect/unstable/rpc"; +import * as Socket from "effect/unstable/socket/Socket"; + +export class TransferHttpRequestError extends Schema.TaggedErrorClass()( + "TransferHttpRequestError", + { + url: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export interface HttpTransferMeasurement { + readonly status: number; + readonly contentEncoding: string | null; + readonly encodedBody: Uint8Array; + readonly encodedBodyBytes: number; + readonly decodedBody: Uint8Array; + readonly decodedBodyBytes: number; + /** HTTP response bytes read from the socket, including status line and headers. */ + readonly wireBytes: number; +} + +export const measureHttpGet = Effect.fn("TransferBudget.measureHttpGet")(function* (input: { + readonly url: string; + readonly headers?: Readonly>; +}) { + return yield* Effect.tryPromise({ + try: () => + new Promise((resolve, reject) => { + let socketBytesBeforeResponse = 0; + const request = NodeHttp.get( + input.url, + { + agent: false, + headers: { + "accept-encoding": "gzip", + connection: "close", + ...input.headers, + }, + }, + (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.once("error", reject); + response.once("end", () => { + try { + const encodedBody = Buffer.concat(chunks); + const header = response.headers["content-encoding"]; + const contentEncoding = Array.isArray(header) + ? (header[0] ?? null) + : (header ?? null); + const decodedBody = + contentEncoding === "gzip" ? NodeZlib.gunzipSync(encodedBody) : encodedBody; + resolve({ + status: response.statusCode ?? 0, + contentEncoding, + encodedBody, + encodedBodyBytes: encodedBody.byteLength, + decodedBody, + decodedBodyBytes: decodedBody.byteLength, + wireBytes: Math.max(0, response.socket.bytesRead - socketBytesBeforeResponse), + }); + } catch (cause) { + reject(cause); + } + }); + }, + ); + request.once("socket", (socket) => { + socketBytesBeforeResponse = socket.bytesRead; + }); + request.once("error", reject); + request.setTimeout(10_000, () => { + request.destroy(new Error(`Timed out reading ${input.url}`)); + }); + }), + catch: (cause) => new TransferHttpRequestError({ url: input.url, cause }), + }); +}); + +export interface WebSocketTransferTotals { + readonly wireBytes: number; + readonly decodedBytes: number; + readonly messages: number; +} + +export interface WebSocketTransferRecorder { + readonly connect: ( + url: string, + protocols: string | string[] | undefined, + cookie: string, + ) => globalThis.WebSocket; + readonly totals: () => WebSocketTransferTotals; + readonly negotiatedExtensions: () => string; +} + +interface NodeWebSocketWithTransport extends NodeSocket.NodeWS.WebSocket { + readonly _socket?: { + readonly bytesRead: number; + }; +} + +function rawDataBytes(data: NodeSocket.NodeWS.RawData): number { + if (Array.isArray(data)) { + return data.reduce((total, chunk) => total + chunk.byteLength, 0); + } + return data.byteLength; +} + +export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder { + let socket: NodeWebSocketWithTransport | null = null; + let decodedBytes = 0; + let messages = 0; + + return { + connect: (url, protocols, cookie) => { + const nextSocket = new NodeSocket.NodeWS.WebSocket(url, protocols, { + headers: { cookie }, + perMessageDeflate: true, + }) as NodeWebSocketWithTransport; + socket = nextSocket; + nextSocket.on("message", (data) => { + const bytes = rawDataBytes(data); + decodedBytes += bytes; + messages += 1; + }); + return nextSocket as unknown as globalThis.WebSocket; + }, + totals: () => ({ + wireBytes: socket?._socket?.bytesRead ?? 0, + decodedBytes, + messages, + }), + negotiatedExtensions: () => socket?.extensions ?? "", + }; +} + +export function transferDelta( + start: WebSocketTransferTotals, + end: WebSocketTransferTotals, +): WebSocketTransferTotals { + return { + wireBytes: Math.max(0, end.wireBytes - start.wireBytes), + decodedBytes: Math.max(0, end.decodedBytes - start.decodedBytes), + messages: Math.max(0, end.messages - start.messages), + }; +} + +export function countingWsRpcProtocolLayer(input: { + readonly url: string; + readonly cookie: string; + readonly recorder: WebSocketTransferRecorder; +}) { + const webSocketConstructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url, protocols) => + input.recorder.connect(url, protocols, input.cookie), + ); + return RpcClient.layerProtocolSocket().pipe( + Layer.provide( + Socket.layerWebSocket(input.url, { openTimeout: "10 seconds" }).pipe( + Layer.provide(webSocketConstructorLayer), + ), + ), + Layer.provide(RpcSerialization.layerJson), + ); +} + +export const makeCountingWsRpcClient = RpcClient.make(WsRpcGroup); +export type CountingWsRpcClient = Effect.Success; diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts index 0e64699de97..095cca4e5e7 100644 --- a/apps/server/integration/TestProviderAdapter.integration.ts +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -11,7 +11,6 @@ import { ProviderDriverKind, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; -import * as Crypto from "effect/Crypto"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; @@ -226,9 +225,9 @@ function missingSessionEffect( export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapterHarnessOptions) => Effect.gen(function* () { const provider = options?.provider ?? ProviderDriverKind.make("codex"); - const crypto = yield* Crypto.Crypto; const runtimeEvents = yield* Queue.unbounded(); let sessionCount = 0; + let eventCount = 0; const sessions = new Map(); const queuedResponsesForNextSession: TestTurnResponse[] = []; const interruptCallsBySession = new Map>(); @@ -242,18 +241,10 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter >(); const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event); - const randomUUIDv4 = (threadId: ThreadId) => - crypto.randomUUIDv4.pipe( - Effect.mapError( - (cause) => - new ProviderAdapterValidationError({ - provider, - operation: "crypto/randomUUIDv4", - issue: `Failed to generate test runtime identifier for thread '${threadId}'.`, - cause, - }), - ), - ); + const nextEventId = (threadId: ThreadId) => { + eventCount += 1; + return EventId.make(`test-provider:${provider}:${threadId}:${eventCount}`); + }; const startSession: ProviderAdapterShape["startSession"] = (input) => Effect.gen(function* () { @@ -322,7 +313,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter for (const fixtureEvent of response.events) { const rawEvent: Record = { ...(fixtureEvent as Record), - eventId: yield* randomUUIDv4(input.threadId), + eventId: nextEventId(input.threadId), provider, sessionId: RuntimeSessionId.make(String(input.threadId)), }; @@ -379,7 +370,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter if (deferredTurnCompletedEvents.length === 0) { yield* emit({ type: "turn.completed", - eventId: EventId.make(yield* randomUUIDv4(input.threadId)), + eventId: nextEventId(input.threadId), provider, createdAt: nowIso(), threadId: state.snapshot.threadId, diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts new file mode 100644 index 00000000000..19b7c150a9e --- /dev/null +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -0,0 +1,206 @@ +import type { ProviderDriverKind } from "@t3tools/contracts"; + +import type { + HttpTransferMeasurement, + WebSocketTransferTotals, +} from "./NetworkTransferMeasurement.integration.ts"; + +export interface TransferBudgetRun { + readonly provider: ProviderDriverKind; + readonly document: HttpTransferMeasurement; + readonly shellSnapshot: HttpTransferMeasurement; + readonly threadSnapshot: HttpTransferMeasurement; + readonly coldOpenWebSocket: WebSocketTransferTotals; + readonly measuredTurnWebSocket: WebSocketTransferTotals; +} + +interface ProviderTransferBudget { + readonly totalWireBytes: number; + readonly documentWireBytes: number; + readonly shellSnapshotWireBytes: number; + readonly threadSnapshotWireBytes: number; + readonly coldOpenWebSocketWireBytes: number; + readonly measuredTurnWebSocketWireBytes: number; + readonly measuredTurnWebSocketDecodedBytes: number; + readonly measuredTurnWebSocketMessages: number; +} + +// These caps leave roughly 30% headroom above the deterministic fixture. The +// CI report preserves the exact values so intentional protocol growth can be +// reviewed and the caps raised explicitly. +export const TRANSFER_BUDGETS: Readonly> = { + codex: { + totalWireBytes: 13_000, + documentWireBytes: 4_000, + shellSnapshotWireBytes: 1_000, + threadSnapshotWireBytes: 3_000, + coldOpenWebSocketWireBytes: 1_600, + measuredTurnWebSocketWireBytes: 3_400, + measuredTurnWebSocketDecodedBytes: 18_000, + measuredTurnWebSocketMessages: 14, + }, + claudeAgent: { + totalWireBytes: 13_000, + documentWireBytes: 4_000, + shellSnapshotWireBytes: 1_000, + threadSnapshotWireBytes: 3_000, + coldOpenWebSocketWireBytes: 1_600, + measuredTurnWebSocketWireBytes: 3_400, + measuredTurnWebSocketDecodedBytes: 18_000, + measuredTurnWebSocketMessages: 14, + }, +}; + +function totalWireBytes(run: TransferBudgetRun): number { + return ( + run.document.wireBytes + + run.shellSnapshot.wireBytes + + run.threadSnapshot.wireBytes + + run.coldOpenWebSocket.wireBytes + + run.measuredTurnWebSocket.wireBytes + ); +} + +function formatBytes(bytes: number): string { + if (bytes < 1_024) return `${bytes} B`; + return `${(bytes / 1_024).toFixed(1)} KiB (${bytes.toLocaleString("en-US")} B)`; +} + +function row( + provider: ProviderDriverKind, + phase: string, + metric: string, + observed: number, + maximum: number, + format: (value: number) => string = formatBytes, +): string { + const status = observed <= maximum ? "PASS" : "FAIL"; + return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | ${format(maximum)} | ${status} |`; +} + +export function transferBudgetViolations(runs: ReadonlyArray): string[] { + const violations: string[] = []; + for (const run of runs) { + const budget = TRANSFER_BUDGETS[run.provider]; + if (!budget) { + violations.push(`${run.provider}: no transfer budget is configured`); + continue; + } + const checks = [ + ["total client-bound wire bytes", totalWireBytes(run), budget.totalWireBytes], + ["document wire bytes", run.document.wireBytes, budget.documentWireBytes], + ["shell snapshot wire bytes", run.shellSnapshot.wireBytes, budget.shellSnapshotWireBytes], + ["thread snapshot wire bytes", run.threadSnapshot.wireBytes, budget.threadSnapshotWireBytes], + [ + "cold-open WebSocket wire bytes", + run.coldOpenWebSocket.wireBytes, + budget.coldOpenWebSocketWireBytes, + ], + [ + "measured-turn WebSocket wire bytes", + run.measuredTurnWebSocket.wireBytes, + budget.measuredTurnWebSocketWireBytes, + ], + [ + "measured-turn WebSocket decoded bytes", + run.measuredTurnWebSocket.decodedBytes, + budget.measuredTurnWebSocketDecodedBytes, + ], + [ + "measured-turn WebSocket messages", + run.measuredTurnWebSocket.messages, + budget.measuredTurnWebSocketMessages, + ], + ] as const; + for (const [metric, observed, maximum] of checks) { + if (observed > maximum) { + violations.push(`${run.provider}: ${metric} was ${observed}, maximum ${maximum}`); + } + } + } + return violations; +} + +export function formatTransferBudgetReport(runs: ReadonlyArray): string { + const lines = [ + "# T3 Code basic-use transfer budget", + "", + "Wire values are client-bound bytes read from local HTTP and WebSocket sockets. They include HTTP response headers and the WebSocket upgrade, but exclude TCP/IP and TLS framing. WebSocket permessage-deflate is negotiated.", + "", + "| Provider | Total client-bound wire | Budget | Result |", + "| --- | ---: | ---: | --- |", + ...runs.flatMap((run) => { + const budget = TRANSFER_BUDGETS[run.provider]; + if (!budget) return []; + const observed = totalWireBytes(run); + return [ + `| ${run.provider} | ${formatBytes(observed)} | ${formatBytes(budget.totalWireBytes)} | ${observed <= budget.totalWireBytes ? "PASS" : "FAIL"} |`, + ]; + }), + "", + "## Detailed measurements", + "", + "| Provider | Phase | Metric | Observed | Budget | Result |", + "| --- | --- | --- | ---: | ---: | --- |", + ]; + + for (const run of runs) { + const budget = TRANSFER_BUDGETS[run.provider]; + if (!budget) continue; + lines.push( + row(run.provider, "document", "HTTP wire", run.document.wireBytes, budget.documentWireBytes), + row( + run.provider, + "shell snapshot", + "HTTP wire", + run.shellSnapshot.wireBytes, + budget.shellSnapshotWireBytes, + ), + row( + run.provider, + "thread snapshot", + "HTTP wire", + run.threadSnapshot.wireBytes, + budget.threadSnapshotWireBytes, + ), + row( + run.provider, + "cold open", + "WebSocket wire", + run.coldOpenWebSocket.wireBytes, + budget.coldOpenWebSocketWireBytes, + ), + row( + run.provider, + "measured turn", + "WebSocket wire", + run.measuredTurnWebSocket.wireBytes, + budget.measuredTurnWebSocketWireBytes, + ), + row( + run.provider, + "measured turn", + "WebSocket decoded", + run.measuredTurnWebSocket.decodedBytes, + budget.measuredTurnWebSocketDecodedBytes, + ), + row( + run.provider, + "measured turn", + "WebSocket messages", + run.measuredTurnWebSocket.messages, + budget.measuredTurnWebSocketMessages, + String, + ), + ); + } + + lines.push("", "## Compression diagnostics", ""); + for (const run of runs) { + lines.push( + `- ${run.provider}: document body ${formatBytes(run.document.encodedBodyBytes)}; shell ${formatBytes(run.shellSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.shellSnapshot.encodedBodyBytes)} gzip; thread ${formatBytes(run.threadSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.threadSnapshot.encodedBodyBytes)} gzip.`, + ); + } + + return `${lines.join("\n")}\n`; +} diff --git a/apps/server/integration/TransferBudgetScenario.integration.ts b/apps/server/integration/TransferBudgetScenario.integration.ts new file mode 100644 index 00000000000..538e8e8d7de --- /dev/null +++ b/apps/server/integration/TransferBudgetScenario.integration.ts @@ -0,0 +1,130 @@ +import { + CommandId, + defaultInstanceIdForDriver, + DEFAULT_MODEL, + DEFAULT_MODEL_BY_PROVIDER, + DEFAULT_PROVIDER_INTERACTION_MODE, + MessageId, + ProjectId, + ProviderDriverKind, + ThreadId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts"; +import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts"; +import { + makeRecordedTransferTurn, + TRANSFER_HISTORY_TURN_COUNT, +} from "./fixtures/transferBudget.ts"; + +export const TRANSFER_PROJECT_ID = ProjectId.make("transfer-budget-project"); +export const TRANSFER_THREAD_ID = ThreadId.make("transfer-budget-thread"); +export const TRANSFER_MEASURED_TURN_INDEX = TRANSFER_HISTORY_TURN_COUNT; + +export function transferModelSelection(provider: ProviderDriverKind) { + return { + instanceId: defaultInstanceIdForDriver(provider), + model: DEFAULT_MODEL_BY_PROVIDER[provider] ?? DEFAULT_MODEL, + }; +} + +function turnTimestamp(turnIndex: number): string { + return `2026-06-01T00:${String(turnIndex).padStart(2, "0")}:00.000Z`; +} + +const waitForTurnQuiesced = Effect.fn("TransferBudget.waitForTurnQuiesced")(function* ( + harness: OrchestrationIntegrationHarness, + checkpointTurnCount: number, +) { + return yield* harness.waitForReceipt( + (receipt): receipt is TurnProcessingQuiescedReceipt => + receipt.type === "turn.processing.quiesced" && + receipt.threadId === TRANSFER_THREAD_ID && + receipt.checkpointTurnCount === checkpointTurnCount, + ); +}); + +export const seedTransferBudgetHistory = Effect.fn("TransferBudget.seedHistory")(function* ( + harness: OrchestrationIntegrationHarness, + provider: ProviderDriverKind, +) { + if (!harness.adapterHarness) { + return yield* Effect.die(new Error("Transfer budget history requires the replay adapter.")); + } + + const modelSelection = transferModelSelection(provider); + yield* harness.engine.dispatch({ + type: "project.create", + commandId: CommandId.make(`transfer:${provider}:project-create`), + projectId: TRANSFER_PROJECT_ID, + title: "Transfer Budget Project", + workspaceRoot: harness.workspaceDir, + defaultModelSelection: modelSelection, + createdAt: turnTimestamp(0), + }); + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make(`transfer:${provider}:thread-create`), + threadId: TRANSFER_THREAD_ID, + projectId: TRANSFER_PROJECT_ID, + title: `${provider} transfer history`, + modelSelection, + runtimeMode: "approval-required", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + branch: "main", + worktreePath: harness.workspaceDir, + createdAt: turnTimestamp(0), + }); + + for (let turnIndex = 0; turnIndex < TRANSFER_HISTORY_TURN_COUNT; turnIndex += 1) { + const response = makeRecordedTransferTurn(provider, turnIndex); + if (turnIndex === 0) { + yield* harness.adapterHarness.queueTurnResponseForNextSession(response); + } else { + yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response); + } + + yield* harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`transfer:${provider}:turn:${turnIndex + 1}`), + threadId: TRANSFER_THREAD_ID, + message: { + messageId: MessageId.make(`transfer-user-${turnIndex + 1}`), + role: "user", + text: `Inspect transfer behavior for historical turn ${turnIndex + 1}.`, + attachments: [], + }, + modelSelection, + runtimeMode: "approval-required", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + createdAt: turnTimestamp(turnIndex), + }); + yield* waitForTurnQuiesced(harness, turnIndex + 1); + } +}); + +export const queueMeasuredTransferTurn = Effect.fn("TransferBudget.queueMeasuredTurn")(function* ( + harness: OrchestrationIntegrationHarness, + provider: ProviderDriverKind, +) { + if (!harness.adapterHarness) { + return yield* Effect.die(new Error("Transfer budget measurement requires the replay adapter.")); + } + yield* harness.adapterHarness.queueTurnResponse( + TRANSFER_THREAD_ID, + makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX), + ); +}); + +export function expectedMeasuredAssistantText(provider: ProviderDriverKind): string { + return makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX) + .events.filter((event) => event.type === "content.delta") + .map((event) => { + const payload = event.payload as { readonly delta?: unknown } | undefined; + return typeof payload?.delta === "string" ? payload.delta : ""; + }) + .join(""); +} + +export { waitForTurnQuiesced }; diff --git a/apps/server/integration/fixtures/transferBudget.ts b/apps/server/integration/fixtures/transferBudget.ts new file mode 100644 index 00000000000..bd192506e54 --- /dev/null +++ b/apps/server/integration/fixtures/transferBudget.ts @@ -0,0 +1,202 @@ +import { EventId, ProviderDriverKind } from "@t3tools/contracts"; + +import type { + FixtureProviderRuntimeEvent, + TestTurnResponse, +} from "../TestProviderAdapter.integration.ts"; + +const FIXTURE_THREAD_ID = "transfer-budget-thread"; +const FIXTURE_TURN_ID = "transfer-budget-turn"; + +const sourceModules = [ + "connection/session.ts", + "connection/supervisor.ts", + "rpc/client.ts", + "rpc/protocol.ts", + "state/shell.ts", + "state/threads.ts", + "state/threadReducer.ts", + "state/shellReducer.ts", + "state/threadSnapshotHttp.ts", + "state/shellSnapshotHttp.ts", + "orchestration/http.ts", + "orchestration/Normalizer.ts", + "orchestration/ActivityPayloadProjection.ts", + "provider/ProviderService.ts", + "provider/ProviderRuntimeIngestion.ts", + "persistence/ProjectionSnapshotQuery.ts", + "persistence/OrchestrationEventStore.ts", + "checkpointing/CheckpointStore.ts", + "checkpointing/CheckpointDiffQuery.ts", + "server.ts", +] as const; + +function fixtureTimestamp(turnIndex: number, eventIndex: number): string { + const minute = String(turnIndex).padStart(2, "0"); + const second = String(eventIndex).padStart(2, "0"); + return `2026-06-01T00:${minute}:${second}.000Z`; +} + +function diagnosticOutput(provider: ProviderDriverKind, turnIndex: number): string { + return sourceModules + .map((modulePath, index) => { + const transferred = 1_024 + turnIndex * 137 + index * 83; + const frames = 2 + ((turnIndex + index) % 7); + const status = index % 5 === 0 ? "reviewed" : index % 3 === 0 ? "updated" : "unchanged"; + return [ + `[${String(index + 1).padStart(2, "0")}] ${modulePath}`, + `provider=${provider} turn=${turnIndex + 1} status=${status}`, + `decodedBytes=${transferred * 3} compressedBytes=${transferred} frames=${frames}`, + `observation=subscription cursor remained monotonic after ${modulePath} emitted its update`, + ].join("\n"); + }) + .join("\n\n"); +} + +function assistantChunks(provider: ProviderDriverKind, turnIndex: number): ReadonlyArray { + const providerName = provider === "codex" ? "Codex" : "Claude"; + return [ + `I traced the ${providerName} request through the environment connection and orchestration layers. `, + `The transfer sample for turn ${turnIndex + 1} includes a command lifecycle, a projected activity, `, + "a checkpoint diff, and the final assistant message. ", + "The shell subscription receives only the latest thread shell while the thread subscription receives ", + "incremental detail events. The existing conversation is not replayed over the socket.\n\n", + "The focused verification keeps snapshot data on gzip-compressed HTTP and resumes both subscriptions ", + "from the returned sequence. This response is intentionally split into the same small canonical text ", + "deltas seen in real provider streams, while orchestration persists one completed assistant message.\n", + ]; +} + +function unifiedDiff(provider: ProviderDriverKind, turnIndex: number): string { + const lines = sourceModules + .slice(0, 8) + .flatMap((modulePath, index) => [ + `diff --git a/${modulePath} b/${modulePath}`, + `--- a/${modulePath}`, + `+++ b/${modulePath}`, + `@@ -${index + 1},2 +${index + 1},3 @@`, + ` const provider = "${provider}";`, + `+const transferTurn = ${turnIndex + 1};`, + `+const transferSample = ${1_500 + index * 97};`, + ]); + return lines.join("\n"); +} + +function baseEvent( + provider: ProviderDriverKind, + turnIndex: number, + eventIndex: number, +): Pick { + return { + eventId: EventId.make(`recorded:${provider}:${turnIndex}:${eventIndex}`), + provider, + createdAt: fixtureTimestamp(turnIndex, eventIndex), + threadId: FIXTURE_THREAD_ID, + }; +} + +/** + * Canonical provider events shaped from real Codex and Claude event logs. The + * content is synthetic so the fixture is safe to commit, while event ordering, + * delta chunking, tool payloads, usage, and diff sizes remain representative. + */ +export function makeRecordedTransferTurn( + provider: ProviderDriverKind, + turnIndex: number, +): TestTurnResponse { + const chunks = assistantChunks(provider, turnIndex); + const events: FixtureProviderRuntimeEvent[] = [ + { + type: "turn.started", + ...baseEvent(provider, turnIndex, 0), + turnId: FIXTURE_TURN_ID, + payload: { + model: provider === "codex" ? "gpt-5.4" : "claude-opus-4-1", + effort: provider === "codex" ? "high" : "default", + }, + }, + { + type: "item.started", + ...baseEvent(provider, turnIndex, 1), + turnId: FIXTURE_TURN_ID, + itemId: `tool-${turnIndex + 1}`, + payload: { + itemType: provider === "codex" ? "command_execution" : "file_change", + status: "inProgress", + title: provider === "codex" ? "Inspect transfer paths" : "Review transfer paths", + detail: "Inspecting the HTTP snapshot and WebSocket projection boundaries.", + }, + }, + { + type: "item.completed", + ...baseEvent(provider, turnIndex, 2), + turnId: FIXTURE_TURN_ID, + itemId: `tool-${turnIndex + 1}`, + payload: { + itemType: provider === "codex" ? "command_execution" : "file_change", + status: "completed", + title: provider === "codex" ? "Inspected transfer paths" : "Reviewed transfer paths", + detail: "Collected a representative multi-module transfer diagnostic.", + data: { + command: provider === "codex" ? "vp test transfer-budget" : "review transfer budget", + exitCode: 0, + aggregatedOutput: diagnosticOutput(provider, turnIndex), + }, + }, + }, + ...chunks.map( + (delta, chunkIndex): FixtureProviderRuntimeEvent => ({ + type: "content.delta", + ...baseEvent(provider, turnIndex, chunkIndex + 3), + turnId: FIXTURE_TURN_ID, + itemId: `assistant-${turnIndex + 1}`, + payload: { + streamKind: "assistant_text", + delta, + contentIndex: chunkIndex, + }, + }), + ), + { + type: "thread.token-usage.updated", + ...baseEvent(provider, turnIndex, chunks.length + 3), + turnId: FIXTURE_TURN_ID, + payload: { + usage: { + usedTokens: 18_000 + turnIndex * 1_900, + maxTokens: 200_000, + inputTokens: 15_000 + turnIndex * 1_700, + cachedInputTokens: 9_000 + turnIndex * 1_100, + outputTokens: 3_000 + turnIndex * 200, + toolUses: 1, + durationMs: 4_000 + turnIndex * 250, + }, + }, + }, + { + type: "turn.diff.updated", + ...baseEvent(provider, turnIndex, chunks.length + 4), + turnId: FIXTURE_TURN_ID, + payload: { + unifiedDiff: unifiedDiff(provider, turnIndex), + }, + }, + { + type: "turn.completed", + ...baseEvent(provider, turnIndex, chunks.length + 5), + turnId: FIXTURE_TURN_ID, + payload: { + state: "completed", + stopReason: "end_turn", + usage: { + inputTokens: 15_000 + turnIndex * 1_700, + outputTokens: 3_000 + turnIndex * 200, + }, + }, + }, + ]; + + return { events }; +} + +export const TRANSFER_HISTORY_TURN_COUNT = 6; diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index a403e228b06..6fccabe917d 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -2,7 +2,8 @@ import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; import * as NodeSocket from "@effect/platform-node/NodeSocket"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as NodeCrypto from "node:crypto"; -import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import * as NodeURL from "node:url"; +import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { AuthAccessTokenType, @@ -16,6 +17,10 @@ import { KeybindingRule, MessageId, ExternalLauncherCommandNotFoundError, + OrchestrationShellSnapshot, + type OrchestrationShellStreamItem, + OrchestrationThreadDetailSnapshot, + type OrchestrationThreadStreamItem, type OrchestrationThreadShell, TerminalNotRunningError, type OrchestrationCommand, @@ -41,6 +46,7 @@ import * as RelayClient from "@t3tools/shared/relayClient"; import { assert, it } from "@effect/vitest"; import { assertFailure, assertInclude, assertTrue } from "@effect/vitest/utils"; import * as Clock from "effect/Clock"; +import * as Config from "effect/Config"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; @@ -52,6 +58,8 @@ import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { ChildProcessSpawner } from "effect/unstable/process"; @@ -70,6 +78,24 @@ import * as Socket from "effect/unstable/socket/Socket"; import { vi } from "vite-plus/test"; const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z"); +const decodeTransferShellSnapshot = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationShellSnapshot), +); +const decodeTransferThreadSnapshot = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationThreadDetailSnapshot), +); + +const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function* ( + queue: Queue.Queue, + predicate: (value: A) => boolean, +) { + const values: A[] = []; + while (true) { + const value = yield* Queue.take(queue); + values.push(value); + if (predicate(value)) return values; + } +}); import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; import * as ServerConfig from "./config.ts"; @@ -123,6 +149,29 @@ import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; import * as Data from "effect/Data"; +import { makeOrchestrationIntegrationHarness } from "../integration/OrchestrationEngineHarness.integration.ts"; +import { + countingWsRpcProtocolLayer, + makeCountingWsRpcClient, + makeWebSocketTransferRecorder, + measureHttpGet, + transferDelta, +} from "../integration/NetworkTransferMeasurement.integration.ts"; +import { + expectedMeasuredAssistantText, + queueMeasuredTransferTurn, + seedTransferBudgetHistory, + TRANSFER_MEASURED_TURN_INDEX, + TRANSFER_THREAD_ID, + transferModelSelection, + waitForTurnQuiesced, +} from "../integration/TransferBudgetScenario.integration.ts"; +import { + formatTransferBudgetReport, + type TransferBudgetRun, + transferBudgetViolations, +} from "../integration/TransferBudgetReport.integration.ts"; + const defaultProjectId = ProjectId.make("project-default"); const defaultThreadId = ThreadId.make("thread-default"); const defaultDesktopBootstrapToken = "test-desktop-bootstrap-token"; @@ -549,9 +598,12 @@ const buildAppUnderTest = (options?: { ), ), ); + const serviceLauncherClientLayer = ServiceLauncherClient.layer.pipe( + Layer.provide(Layer.succeed(HostProcessEnvironment, {})), + ); const servedRoutesLayer = HttpRouter.serve( - makeRoutesLayer.pipe(Layer.provide(ServiceLauncherClient.layer)), + makeRoutesLayer.pipe(Layer.provide(serviceLauncherClientLayer)), { disableListenLog: true, disableLogger: true, @@ -1319,6 +1371,28 @@ const getWsServerUrl = ( ); }); +// Mirrors NodeHttpServer.layerTest, which does not expose server options, +// with the production `websocket: { perMessageDeflate: true }` setting. +const NodeHttpServerTestWithWsDeflate = HttpServer.layerTestClient.pipe( + Layer.provide( + Layer.fresh(FetchHttpClient.layer).pipe( + Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false })), + ), + ), + Layer.provideMerge( + Layer.unwrap( + Effect.map( + Effect.promise(() => import("node:http")), + (NodeHttp) => + NodeHttpServer.layer(NodeHttp.createServer, { + port: 0, + websocket: { perMessageDeflate: true }, + }), + ), + ), + ), +); + it.layer(NodeServices.layer)("server router seam", (it) => { it.effect("parks HTTP ingress until command readiness", () => Effect.gen(function* () { @@ -3219,28 +3293,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); - // Mirrors NodeHttpServer.layerTest, which does not expose server options, - // with the production `websocket: { perMessageDeflate: true }` setting. - const NodeHttpServerTestWithWsDeflate = HttpServer.layerTestClient.pipe( - Layer.provide( - Layer.fresh(FetchHttpClient.layer).pipe( - Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false })), - ), - ), - Layer.provideMerge( - Layer.unwrap( - Effect.map( - Effect.promise(() => import("node:http")), - (NodeHttp) => - NodeHttpServer.layer(NodeHttp.createServer, { - port: 0, - websocket: { perMessageDeflate: true }, - }), - ), - ), - ), - ); - it.effect("negotiates permessage-deflate with clients that offer it", () => Effect.gen(function* () { yield* buildAppUnderTest(); @@ -7839,3 +7891,179 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); }); + +it.live("reports basic-use HTTP and WebSocket transfer budgets", () => + Effect.gen(function* () { + const providers = [ + ProviderDriverKind.make("codex"), + ProviderDriverKind.make("claudeAgent"), + ] as const; + const staticDir = NodeURL.fileURLToPath(new URL("../../web", import.meta.url)); + + const runs = yield* Effect.forEach( + providers, + (provider) => + Effect.acquireUseRelease( + makeOrchestrationIntegrationHarness({ provider }), + (harness) => + Effect.gen(function* () { + yield* seedTransferBudgetHistory(harness, provider); + yield* buildAppUnderTest({ + config: { staticDir }, + layers: { + orchestrationEngine: harness.engine, + projectionSnapshotQuery: harness.snapshotQuery, + }, + }); + + const baseUrl = yield* getHttpServerUrl(); + const cookie = yield* getAuthenticatedSessionCookieHeader(); + const document = yield* measureHttpGet({ url: `${baseUrl}/` }); + assert.equal(document.status, 200); + + const recorder = makeWebSocketTransferRecorder(); + const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws"; + const protocolLayer = countingWsRpcProtocolLayer({ + url: wsUrl, + cookie, + recorder, + }); + + return yield* Effect.scoped( + Effect.gen(function* () { + const client = yield* makeCountingWsRpcClient; + yield* client[WS_METHODS.serverGetConfig]({}); + + const configEvents = yield* Queue.unbounded(); + yield* client[WS_METHODS.subscribeServerConfig]({}).pipe( + Stream.runForEach((event) => + Queue.offer(configEvents, event).pipe(Effect.asVoid), + ), + Effect.forkScoped, + ); + yield* Queue.take(configEvents); + + const shellSnapshot = yield* measureHttpGet({ + url: `${baseUrl}/api/orchestration/shell`, + headers: { cookie }, + }); + assert.equal(shellSnapshot.status, 200); + assert.equal(shellSnapshot.contentEncoding, "gzip"); + const decodedShell = yield* decodeTransferShellSnapshot( + Buffer.from(shellSnapshot.decodedBody).toString("utf8"), + ); + + const shellItems = yield* Queue.unbounded(); + yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: decodedShell.snapshotSequence, + requestCompletionMarker: true, + }).pipe( + Stream.runForEach((item) => Queue.offer(shellItems, item).pipe(Effect.asVoid)), + Effect.forkScoped, + ); + const initialShellItems = yield* collectQueueUntil( + shellItems, + (item) => item.kind === "synchronized", + ); + assert.isFalse(initialShellItems.some((item) => item.kind === "snapshot")); + + const threadSnapshot = yield* measureHttpGet({ + url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`, + headers: { cookie }, + }); + assert.equal(threadSnapshot.status, 200); + assert.equal(threadSnapshot.contentEncoding, "gzip"); + const decodedThread = yield* decodeTransferThreadSnapshot( + Buffer.from(threadSnapshot.decodedBody).toString("utf8"), + ); + assert.equal(decodedThread.thread.messages.length, 12); + + const threadItems = yield* Queue.unbounded(); + yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: TRANSFER_THREAD_ID, + afterSequence: decodedThread.snapshotSequence, + requestCompletionMarker: true, + }).pipe( + Stream.runForEach((item) => Queue.offer(threadItems, item).pipe(Effect.asVoid)), + Effect.forkScoped, + ); + const initialThreadItems = yield* collectQueueUntil( + threadItems, + (item) => item.kind === "synchronized", + ); + assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot")); + assert.include(recorder.negotiatedExtensions(), "permessage-deflate"); + + const coldOpenWebSocket = recorder.totals(); + yield* queueMeasuredTransferTurn(harness, provider); + const turnStartTotals = recorder.totals(); + yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make(`transfer:${provider}:measured-turn`), + threadId: TRANSFER_THREAD_ID, + message: { + messageId: MessageId.make("transfer-user-measured"), + role: "user", + text: "Measure the client-bound transfer for this turn.", + attachments: [], + }, + modelSelection: transferModelSelection(provider), + runtimeMode: "approval-required", + interactionMode: "default", + createdAt: "2026-06-01T00:06:00.000Z", + }); + yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1); + const finalSequence = yield* harness.engine.latestSequence; + + yield* collectQueueUntil( + threadItems, + (item) => item.kind === "event" && item.event.sequence >= finalSequence, + ); + yield* collectQueueUntil( + shellItems, + (item) => + item.kind !== "snapshot" && + item.kind !== "synchronized" && + item.sequence >= finalSequence, + ); + const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals()); + + const finalThreadSnapshot = yield* harness.snapshotQuery + .getThreadDetailSnapshot(TRANSFER_THREAD_ID) + .pipe(Effect.map(Option.getOrThrow)); + const finalAssistant = finalThreadSnapshot.thread.messages.findLast( + (message) => message.role === "assistant", + ); + assert.equal(finalAssistant?.text, expectedMeasuredAssistantText(provider)); + assert.equal(finalAssistant?.streaming, false); + assert.equal(finalThreadSnapshot.thread.session?.status, "ready"); + assert.equal(finalThreadSnapshot.thread.checkpoints.length, 7); + + return { + provider, + document, + shellSnapshot, + threadSnapshot, + coldOpenWebSocket, + measuredTurnWebSocket, + } satisfies TransferBudgetRun; + }).pipe(Effect.provide(protocolLayer)), + ); + }), + (harness) => harness.dispose, + ).pipe(Effect.provide(NodeHttpServerTestWithWsDeflate)), + { concurrency: 1 }, + ); + + const report = formatTransferBudgetReport(runs); + yield* Effect.logInfo(`\n${report}`); + const reportPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_REPORT_PATH").pipe( + Config.option, + ); + if (Option.isSome(reportPath)) { + const fileSystem = yield* FileSystem.FileSystem; + yield* fileSystem.writeFileString(reportPath.value, report); + } + assert.deepEqual(transferBudgetViolations(runs), []); + }).pipe(Effect.provide(NodeServices.layer)), +); From 6b0a47d892eae5cfe5af2a2189a2f261e2a7ea1b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 12:05:34 -0700 Subject: [PATCH 02/10] test(server): stress transfer fixture with heavy history --- .../OrchestrationEngineHarness.integration.ts | 7 + .../TransferBudgetReport.integration.ts | 37 +- .../TransferBudgetScenario.integration.ts | 23 +- .../integration/fixtures/transferBudget.ts | 333 +++++++++++++---- apps/server/src/server.test.ts | 352 +++++++++--------- 5 files changed, 485 insertions(+), 267 deletions(-) diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index c3f77d677b1..8daa9555655 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -55,6 +55,7 @@ import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceip import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts"; import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts"; +import { ProviderRuntimeIngestionService } from "../src/orchestration/Services/ProviderRuntimeIngestion.ts"; import { OrchestrationEngineService, type OrchestrationEngineShape, @@ -218,6 +219,7 @@ export interface OrchestrationIntegrationHarness { timeoutMs?: number, ): Effect.Effect; }; + readonly drainProviderRuntime: Effect.Effect; readonly dispose: Effect.Effect; } @@ -392,6 +394,10 @@ export const makeOrchestrationIntegrationHarness = ( const reactor = yield* tryRuntimePromise("load OrchestrationReactor service", () => runtime.runPromise(Effect.service(OrchestrationReactor)), ).pipe(Effect.orDie); + const providerRuntimeIngestion = yield* tryRuntimePromise( + "load ProviderRuntimeIngestion service", + () => runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)), + ).pipe(Effect.orDie); const snapshotQuery = yield* tryRuntimePromise("load ProjectionSnapshotQuery service", () => runtime.runPromise(Effect.service(ProjectionSnapshotQuery)), ).pipe(Effect.orDie); @@ -556,6 +562,7 @@ export const makeOrchestrationIntegrationHarness = ( waitForDomainEvent, waitForPendingApproval, waitForReceipt, + drainProviderRuntime: providerRuntimeIngestion.drain, dispose, } satisfies OrchestrationIntegrationHarness; }); diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 19b7c150a9e..4b3251b713a 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -4,6 +4,13 @@ import type { HttpTransferMeasurement, WebSocketTransferTotals, } from "./NetworkTransferMeasurement.integration.ts"; +import { + TRANSFER_HISTORY_MCP_RESULT_BYTES, + TRANSFER_HISTORY_TOOLS_PER_TURN, + TRANSFER_HISTORY_TURN_COUNT, + TRANSFER_MEASURED_MCP_RESULT_BYTES, + TRANSFER_MEASURED_TOOLS, +} from "./fixtures/transferBudget.ts"; export interface TransferBudgetRun { readonly provider: ProviderDriverKind; @@ -30,24 +37,24 @@ interface ProviderTransferBudget { // reviewed and the caps raised explicitly. export const TRANSFER_BUDGETS: Readonly> = { codex: { - totalWireBytes: 13_000, - documentWireBytes: 4_000, + totalWireBytes: 2_900_000, + documentWireBytes: 1_700, shellSnapshotWireBytes: 1_000, - threadSnapshotWireBytes: 3_000, + threadSnapshotWireBytes: 2_550_000, coldOpenWebSocketWireBytes: 1_600, - measuredTurnWebSocketWireBytes: 3_400, - measuredTurnWebSocketDecodedBytes: 18_000, - measuredTurnWebSocketMessages: 14, + measuredTurnWebSocketWireBytes: 360_000, + measuredTurnWebSocketDecodedBytes: 2_050_000, + measuredTurnWebSocketMessages: 550, }, claudeAgent: { - totalWireBytes: 13_000, - documentWireBytes: 4_000, + totalWireBytes: 2_900_000, + documentWireBytes: 1_700, shellSnapshotWireBytes: 1_000, - threadSnapshotWireBytes: 3_000, + threadSnapshotWireBytes: 2_550_000, coldOpenWebSocketWireBytes: 1_600, - measuredTurnWebSocketWireBytes: 3_400, - measuredTurnWebSocketDecodedBytes: 18_000, - measuredTurnWebSocketMessages: 14, + measuredTurnWebSocketWireBytes: 360_000, + measuredTurnWebSocketDecodedBytes: 2_050_000, + measuredTurnWebSocketMessages: 550, }, }; @@ -63,6 +70,9 @@ function totalWireBytes(run: TransferBudgetRun): number { function formatBytes(bytes: number): string { if (bytes < 1_024) return `${bytes} B`; + if (bytes >= 1_024 * 1_024) { + return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB (${bytes.toLocaleString("en-US")} B)`; + } return `${(bytes / 1_024).toFixed(1)} KiB (${bytes.toLocaleString("en-US")} B)`; } @@ -123,9 +133,10 @@ export function transferBudgetViolations(runs: ReadonlyArray) export function formatTransferBudgetReport(runs: ReadonlyArray): string { const lines = [ - "# T3 Code basic-use transfer budget", + "# T3 Code stress transfer budget", "", "Wire values are client-bound bytes read from local HTTP and WebSocket sockets. They include HTTP response headers and the WebSocket upgrade, but exclude TCP/IP and TLS framing. WebSocket permessage-deflate is negotiated.", + `Scenario: ${TRANSFER_HISTORY_TURN_COUNT} historical turns with ${TRANSFER_HISTORY_TOOLS_PER_TURN} command tools and one retained ${formatBytes(TRANSFER_HISTORY_MCP_RESULT_BYTES)} MCP result each, followed by one measured turn with ${TRANSFER_MEASURED_TOOLS} command tools and a retained ${formatBytes(TRANSFER_MEASURED_MCP_RESULT_BYTES)} MCP result. Payload sizes are calibrated from heavy local Codex and Claude histories and contain no user data.`, "", "| Provider | Total client-bound wire | Budget | Result |", "| --- | ---: | ---: | --- |", diff --git a/apps/server/integration/TransferBudgetScenario.integration.ts b/apps/server/integration/TransferBudgetScenario.integration.ts index 538e8e8d7de..0807481abd1 100644 --- a/apps/server/integration/TransferBudgetScenario.integration.ts +++ b/apps/server/integration/TransferBudgetScenario.integration.ts @@ -14,6 +14,7 @@ import * as Effect from "effect/Effect"; import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts"; import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts"; import { + expectedRecordedAssistantText, makeRecordedTransferTurn, TRANSFER_HISTORY_TURN_COUNT, } from "./fixtures/transferBudget.ts"; @@ -33,16 +34,20 @@ function turnTimestamp(turnIndex: number): string { return `2026-06-01T00:${String(turnIndex).padStart(2, "0")}:00.000Z`; } +export const TRANSFER_MEASURED_TURN_CREATED_AT = turnTimestamp(TRANSFER_MEASURED_TURN_INDEX); + const waitForTurnQuiesced = Effect.fn("TransferBudget.waitForTurnQuiesced")(function* ( harness: OrchestrationIntegrationHarness, checkpointTurnCount: number, ) { - return yield* harness.waitForReceipt( + const receipt = yield* harness.waitForReceipt( (receipt): receipt is TurnProcessingQuiescedReceipt => receipt.type === "turn.processing.quiesced" && receipt.threadId === TRANSFER_THREAD_ID && receipt.checkpointTurnCount === checkpointTurnCount, ); + yield* harness.drainProviderRuntime; + return receipt; }); export const seedTransferBudgetHistory = Effect.fn("TransferBudget.seedHistory")(function* ( @@ -111,20 +116,12 @@ export const queueMeasuredTransferTurn = Effect.fn("TransferBudget.queueMeasured if (!harness.adapterHarness) { return yield* Effect.die(new Error("Transfer budget measurement requires the replay adapter.")); } - yield* harness.adapterHarness.queueTurnResponse( - TRANSFER_THREAD_ID, - makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX), - ); + const response = makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX); + yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response); }); export function expectedMeasuredAssistantText(provider: ProviderDriverKind): string { - return makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX) - .events.filter((event) => event.type === "content.delta") - .map((event) => { - const payload = event.payload as { readonly delta?: unknown } | undefined; - return typeof payload?.delta === "string" ? payload.delta : ""; - }) - .join(""); + return expectedRecordedAssistantText(provider, TRANSFER_MEASURED_TURN_INDEX); } -export { waitForTurnQuiesced }; +export { TRANSFER_HISTORY_TURN_COUNT, waitForTurnQuiesced }; diff --git a/apps/server/integration/fixtures/transferBudget.ts b/apps/server/integration/fixtures/transferBudget.ts index bd192506e54..dec986a2590 100644 --- a/apps/server/integration/fixtures/transferBudget.ts +++ b/apps/server/integration/fixtures/transferBudget.ts @@ -8,6 +8,12 @@ import type { const FIXTURE_THREAD_ID = "transfer-budget-thread"; const FIXTURE_TURN_ID = "transfer-budget-turn"; +export const TRANSFER_HISTORY_TURN_COUNT = 10; +export const TRANSFER_HISTORY_TOOLS_PER_TURN = 100; +export const TRANSFER_MEASURED_TOOLS = 150; +export const TRANSFER_HISTORY_MCP_RESULT_BYTES = 900_000; +export const TRANSFER_MEASURED_MCP_RESULT_BYTES = 1_100_000; + const sourceModules = [ "connection/session.ts", "connection/supervisor.ts", @@ -33,38 +39,94 @@ const sourceModules = [ function fixtureTimestamp(turnIndex: number, eventIndex: number): string { const minute = String(turnIndex).padStart(2, "0"); - const second = String(eventIndex).padStart(2, "0"); - return `2026-06-01T00:${minute}:${second}.000Z`; + const second = String(Math.floor(eventIndex / 1_000)).padStart(2, "0"); + const millisecond = String(eventIndex % 1_000).padStart(3, "0"); + return `2026-06-01T00:${minute}:${second}.${millisecond}Z`; +} + +function mix(value: number): number { + let mixed = value | 0; + mixed ^= mixed >>> 16; + mixed = Math.imul(mixed, 0x7feb352d); + mixed ^= mixed >>> 15; + mixed = Math.imul(mixed, 0x846ca68b); + mixed ^= mixed >>> 16; + return mixed >>> 0; +} + +function digest(seed: number): string { + return [0, 1, 2, 3] + .map((offset) => + mix(seed + offset * 0x9e3779b9) + .toString(16) + .padStart(8, "0"), + ) + .join(""); } -function diagnosticOutput(provider: ProviderDriverKind, turnIndex: number): string { - return sourceModules - .map((modulePath, index) => { - const transferred = 1_024 + turnIndex * 137 + index * 83; - const frames = 2 + ((turnIndex + index) % 7); - const status = index % 5 === 0 ? "reviewed" : index % 3 === 0 ? "updated" : "unchanged"; - return [ - `[${String(index + 1).padStart(2, "0")}] ${modulePath}`, - `provider=${provider} turn=${turnIndex + 1} status=${status}`, - `decodedBytes=${transferred * 3} compressedBytes=${transferred} frames=${frames}`, - `observation=subscription cursor remained monotonic after ${modulePath} emitted its update`, - ].join("\n"); - }) - .join("\n\n"); +/** Produces safe, deterministic output with enough entropy to exercise gzip. */ +function diagnosticOutput(input: { + readonly provider: ProviderDriverKind; + readonly turnIndex: number; + readonly toolIndex: number; + readonly targetBytes: number; +}): string { + const chunks: string[] = []; + const providerSeed = input.provider === "codex" ? 0x43_4f_44_45 : 0x43_4c_41_55; + let length = 0; + let lineIndex = 0; + + while (length < input.targetBytes) { + const modulePath = sourceModules[(input.toolIndex + lineIndex) % sourceModules.length]; + const seed = + providerSeed + input.turnIndex * 100_003 + input.toolIndex * 10_007 + lineIndex * 101; + const line = + `${String(lineIndex + 1).padStart(6, "0")} ${modulePath} ` + + `operation=project-transfer-${input.turnIndex + 1}-${input.toolIndex + 1} ` + + `cursor=${mix(seed)} digest=${digest(seed)} status=completed\n`; + chunks.push(line); + length += line.length; + lineIndex += 1; + } + + return chunks.join("").slice(0, input.targetBytes); +} + +function toolOutputBytes(toolIndex: number, measuredTurn: boolean): number { + if (measuredTurn) { + if (toolIndex >= 136) return 35_000; + if (toolIndex >= 101) return 8_000; + return 1_000; + } + + return 1_000; } function assistantChunks(provider: ProviderDriverKind, turnIndex: number): ReadonlyArray { const providerName = provider === "codex" ? "Codex" : "Claude"; - return [ + const paragraphs: string[] = [ `I traced the ${providerName} request through the environment connection and orchestration layers. `, - `The transfer sample for turn ${turnIndex + 1} includes a command lifecycle, a projected activity, `, - "a checkpoint diff, and the final assistant message. ", - "The shell subscription receives only the latest thread shell while the thread subscription receives ", - "incremental detail events. The existing conversation is not replayed over the socket.\n\n", - "The focused verification keeps snapshot data on gzip-compressed HTTP and resumes both subscriptions ", - "from the returned sequence. This response is intentionally split into the same small canonical text ", - "deltas seen in real provider streams, while orchestration persists one completed assistant message.\n", ]; + let paragraphIndex = 0; + while (paragraphs.join("").length < 4_096) { + const modulePath = sourceModules[paragraphIndex % sourceModules.length]; + paragraphs.push( + `Pass ${paragraphIndex + 1} reviewed ${modulePath} for turn ${turnIndex + 1}. ` + + "The shell cursor stayed monotonic, the thread snapshot remained resumable, and the client received only incremental events. ", + ); + paragraphIndex += 1; + } + const text = paragraphs.join("").slice(0, 4_096); + return Array.from({ length: Math.ceil(text.length / 256) }, (_, index) => + text.slice(index * 256, (index + 1) * 256), + ); +} + +export function expectedRecordedAssistantText( + provider: ProviderDriverKind, + turnIndex: number, +): string { + return assistantChunks(provider, turnIndex).join(""); } function unifiedDiff(provider: ProviderDriverKind, turnIndex: number): string { @@ -96,71 +158,194 @@ function baseEvent( } /** - * Canonical provider events shaped from real Codex and Claude event logs. The - * content is synthetic so the fixture is safe to commit, while event ordering, - * delta chunking, tool payloads, usage, and diff sizes remain representative. + * Synthetic canonical events calibrated from heavy local Codex and Claude + * threads. Ten historical turns produce roughly 2,000 command activity events + * plus 9 MB of retained MCP results without committing user content. Command + * output is intentionally modest because the client projection strips it. */ export function makeRecordedTransferTurn( provider: ProviderDriverKind, turnIndex: number, ): TestTurnResponse { - const chunks = assistantChunks(provider, turnIndex); - const events: FixtureProviderRuntimeEvent[] = [ - { - type: "turn.started", - ...baseEvent(provider, turnIndex, 0), - turnId: FIXTURE_TURN_ID, - payload: { - model: provider === "codex" ? "gpt-5.4" : "claude-opus-4-1", - effort: provider === "codex" ? "high" : "default", - }, + const measuredTurn = turnIndex >= TRANSFER_HISTORY_TURN_COUNT; + const toolCount = measuredTurn ? TRANSFER_MEASURED_TOOLS : TRANSFER_HISTORY_TOOLS_PER_TURN; + const turnId = `${FIXTURE_TURN_ID}-${turnIndex + 1}`; + const events: FixtureProviderRuntimeEvent[] = []; + let eventIndex = 0; + + events.push({ + type: "turn.started", + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, + payload: { + model: provider === "codex" ? "gpt-5.4" : "claude-opus-4-1", + effort: provider === "codex" ? "high" : "default", }, + }); + + for (let toolIndex = 0; toolIndex < toolCount; toolIndex += 1) { + const itemId = `tool-${turnIndex + 1}-${toolIndex + 1}`; + const command = + provider === "codex" + ? `vp test transfer-budget-${toolIndex + 1}` + : `review transfer budget ${toolIndex + 1}`; + events.push( + { + type: "item.started", + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, + itemId, + payload: { + itemType: "command_execution", + status: "inProgress", + title: `Inspect transfer path ${toolIndex + 1}`, + detail: "Inspecting the HTTP snapshot and WebSocket projection boundaries.", + data: { + threadId: FIXTURE_THREAD_ID, + turnId, + startedAtMs: turnIndex * 60_000 + eventIndex, + item: { + id: itemId, + type: "commandExecution", + command, + cwd: "/workspace/transfer-budget", + processId: String(toolIndex + 1), + status: "inProgress", + commandActions: [], + aggregatedOutput: "", + }, + }, + }, + }, + { + type: "item.completed", + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, + itemId, + payload: { + itemType: "command_execution", + status: "completed", + title: `Inspected transfer path ${toolIndex + 1}`, + detail: "Collected a deterministic multi-module transfer diagnostic.", + data: { + threadId: FIXTURE_THREAD_ID, + turnId, + completedAtMs: turnIndex * 60_000 + eventIndex, + item: { + id: itemId, + type: "commandExecution", + command, + cwd: "/workspace/transfer-budget", + processId: String(toolIndex + 1), + status: "completed", + commandActions: [], + aggregatedOutput: diagnosticOutput({ + provider, + turnIndex, + toolIndex, + targetBytes: toolOutputBytes(toolIndex, measuredTurn), + }), + exitCode: 0, + durationMs: 500 + toolIndex, + }, + }, + }, + }, + ); + } + + const mcpItemId = `mcp-${turnIndex + 1}`; + const mcpResultBytes = measuredTurn + ? TRANSFER_MEASURED_MCP_RESULT_BYTES + : TRANSFER_HISTORY_MCP_RESULT_BYTES; + events.push( { type: "item.started", - ...baseEvent(provider, turnIndex, 1), - turnId: FIXTURE_TURN_ID, - itemId: `tool-${turnIndex + 1}`, + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, + itemId: mcpItemId, payload: { - itemType: provider === "codex" ? "command_execution" : "file_change", + itemType: "mcp_tool_call", status: "inProgress", - title: provider === "codex" ? "Inspect transfer paths" : "Review transfer paths", - detail: "Inspecting the HTTP snapshot and WebSocket projection boundaries.", + title: "fixture-history · inspect_transfer_log", + detail: "Reading a retained diagnostic result from the provider history.", + data: { + startedAtMs: turnIndex * 60_000 + eventIndex, + threadId: FIXTURE_THREAD_ID, + turnId, + item: { + type: "mcpToolCall", + id: mcpItemId, + server: "fixture-history", + tool: "inspect_transfer_log", + arguments: { turn: turnIndex + 1 }, + status: "inProgress", + }, + }, }, }, { type: "item.completed", - ...baseEvent(provider, turnIndex, 2), - turnId: FIXTURE_TURN_ID, - itemId: `tool-${turnIndex + 1}`, + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, + itemId: mcpItemId, payload: { - itemType: provider === "codex" ? "command_execution" : "file_change", + itemType: "mcp_tool_call", status: "completed", - title: provider === "codex" ? "Inspected transfer paths" : "Reviewed transfer paths", - detail: "Collected a representative multi-module transfer diagnostic.", + title: "fixture-history · inspect_transfer_log", + detail: "Retained a deterministic diagnostic result in the thread history.", data: { - command: provider === "codex" ? "vp test transfer-budget" : "review transfer budget", - exitCode: 0, - aggregatedOutput: diagnosticOutput(provider, turnIndex), + completedAtMs: turnIndex * 60_000 + eventIndex, + threadId: FIXTURE_THREAD_ID, + turnId, + item: { + type: "mcpToolCall", + id: mcpItemId, + server: "fixture-history", + tool: "inspect_transfer_log", + arguments: { turn: turnIndex + 1 }, + durationMs: 1_000 + turnIndex, + error: null, + result: { + content: [ + { + type: "text", + text: diagnosticOutput({ + provider, + turnIndex, + toolIndex: toolCount, + targetBytes: mcpResultBytes, + }), + }, + ], + }, + status: "completed", + }, }, }, }, - ...chunks.map( - (delta, chunkIndex): FixtureProviderRuntimeEvent => ({ - type: "content.delta", - ...baseEvent(provider, turnIndex, chunkIndex + 3), - turnId: FIXTURE_TURN_ID, - itemId: `assistant-${turnIndex + 1}`, - payload: { - streamKind: "assistant_text", - delta, - contentIndex: chunkIndex, - }, - }), - ), + ); + + const chunks = assistantChunks(provider, turnIndex); + for (const [contentIndex, delta] of chunks.entries()) { + events.push({ + type: "content.delta", + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, + itemId: `assistant-${turnIndex + 1}`, + payload: { + streamKind: "assistant_text", + delta, + contentIndex, + }, + }); + } + + events.push( { type: "thread.token-usage.updated", - ...baseEvent(provider, turnIndex, chunks.length + 3), - turnId: FIXTURE_TURN_ID, + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, payload: { usage: { usedTokens: 18_000 + turnIndex * 1_900, @@ -168,23 +353,23 @@ export function makeRecordedTransferTurn( inputTokens: 15_000 + turnIndex * 1_700, cachedInputTokens: 9_000 + turnIndex * 1_100, outputTokens: 3_000 + turnIndex * 200, - toolUses: 1, + toolUses: toolCount, durationMs: 4_000 + turnIndex * 250, }, }, }, { type: "turn.diff.updated", - ...baseEvent(provider, turnIndex, chunks.length + 4), - turnId: FIXTURE_TURN_ID, + ...baseEvent(provider, turnIndex, eventIndex++), + turnId, payload: { unifiedDiff: unifiedDiff(provider, turnIndex), }, }, { type: "turn.completed", - ...baseEvent(provider, turnIndex, chunks.length + 5), - turnId: FIXTURE_TURN_ID, + ...baseEvent(provider, turnIndex, eventIndex), + turnId, payload: { state: "completed", stopReason: "end_turn", @@ -194,9 +379,7 @@ export function makeRecordedTransferTurn( }, }, }, - ]; + ); return { events }; } - -export const TRANSFER_HISTORY_TURN_COUNT = 6; diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 6fccabe917d..671f0080a2e 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -161,6 +161,8 @@ import { expectedMeasuredAssistantText, queueMeasuredTransferTurn, seedTransferBudgetHistory, + TRANSFER_HISTORY_TURN_COUNT, + TRANSFER_MEASURED_TURN_CREATED_AT, TRANSFER_MEASURED_TURN_INDEX, TRANSFER_THREAD_ID, transferModelSelection, @@ -7892,178 +7894,196 @@ it.layer(NodeServices.layer)("server router seam", (it) => { ); }); -it.live("reports basic-use HTTP and WebSocket transfer budgets", () => - Effect.gen(function* () { - const providers = [ - ProviderDriverKind.make("codex"), - ProviderDriverKind.make("claudeAgent"), - ] as const; - const staticDir = NodeURL.fileURLToPath(new URL("../../web", import.meta.url)); - - const runs = yield* Effect.forEach( - providers, - (provider) => - Effect.acquireUseRelease( - makeOrchestrationIntegrationHarness({ provider }), - (harness) => - Effect.gen(function* () { - yield* seedTransferBudgetHistory(harness, provider); - yield* buildAppUnderTest({ - config: { staticDir }, - layers: { - orchestrationEngine: harness.engine, - projectionSnapshotQuery: harness.snapshotQuery, - }, - }); +it.live( + "reports stress HTTP and WebSocket transfer budgets", + () => + Effect.gen(function* () { + const providers = [ + ProviderDriverKind.make("codex"), + ProviderDriverKind.make("claudeAgent"), + ] as const; + const staticDir = NodeURL.fileURLToPath(new URL("../../web", import.meta.url)); + + const runs = yield* Effect.forEach( + providers, + (provider) => + Effect.acquireUseRelease( + makeOrchestrationIntegrationHarness({ provider }), + (harness) => + Effect.gen(function* () { + yield* seedTransferBudgetHistory(harness, provider); + yield* buildAppUnderTest({ + config: { staticDir }, + layers: { + orchestrationEngine: harness.engine, + projectionSnapshotQuery: harness.snapshotQuery, + }, + }); - const baseUrl = yield* getHttpServerUrl(); - const cookie = yield* getAuthenticatedSessionCookieHeader(); - const document = yield* measureHttpGet({ url: `${baseUrl}/` }); - assert.equal(document.status, 200); - - const recorder = makeWebSocketTransferRecorder(); - const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws"; - const protocolLayer = countingWsRpcProtocolLayer({ - url: wsUrl, - cookie, - recorder, - }); + const baseUrl = yield* getHttpServerUrl(); + const cookie = yield* getAuthenticatedSessionCookieHeader(); + const document = yield* measureHttpGet({ url: `${baseUrl}/` }); + assert.equal(document.status, 200); + + const recorder = makeWebSocketTransferRecorder(); + const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws"; + const protocolLayer = countingWsRpcProtocolLayer({ + url: wsUrl, + cookie, + recorder, + }); - return yield* Effect.scoped( - Effect.gen(function* () { - const client = yield* makeCountingWsRpcClient; - yield* client[WS_METHODS.serverGetConfig]({}); - - const configEvents = yield* Queue.unbounded(); - yield* client[WS_METHODS.subscribeServerConfig]({}).pipe( - Stream.runForEach((event) => - Queue.offer(configEvents, event).pipe(Effect.asVoid), - ), - Effect.forkScoped, - ); - yield* Queue.take(configEvents); + return yield* Effect.scoped( + Effect.gen(function* () { + const client = yield* makeCountingWsRpcClient; + yield* client[WS_METHODS.serverGetConfig]({}); + + const configEvents = yield* Queue.unbounded(); + yield* client[WS_METHODS.subscribeServerConfig]({}).pipe( + Stream.runForEach((event) => + Queue.offer(configEvents, event).pipe(Effect.asVoid), + ), + Effect.forkScoped, + ); + yield* Queue.take(configEvents); - const shellSnapshot = yield* measureHttpGet({ - url: `${baseUrl}/api/orchestration/shell`, - headers: { cookie }, - }); - assert.equal(shellSnapshot.status, 200); - assert.equal(shellSnapshot.contentEncoding, "gzip"); - const decodedShell = yield* decodeTransferShellSnapshot( - Buffer.from(shellSnapshot.decodedBody).toString("utf8"), - ); + const shellSnapshot = yield* measureHttpGet({ + url: `${baseUrl}/api/orchestration/shell`, + headers: { cookie }, + }); + assert.equal(shellSnapshot.status, 200); + assert.equal(shellSnapshot.contentEncoding, "gzip"); + const decodedShell = yield* decodeTransferShellSnapshot( + Buffer.from(shellSnapshot.decodedBody).toString("utf8"), + ); - const shellItems = yield* Queue.unbounded(); - yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({ - afterSequence: decodedShell.snapshotSequence, - requestCompletionMarker: true, - }).pipe( - Stream.runForEach((item) => Queue.offer(shellItems, item).pipe(Effect.asVoid)), - Effect.forkScoped, - ); - const initialShellItems = yield* collectQueueUntil( - shellItems, - (item) => item.kind === "synchronized", - ); - assert.isFalse(initialShellItems.some((item) => item.kind === "snapshot")); + const shellItems = yield* Queue.unbounded(); + yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: decodedShell.snapshotSequence, + requestCompletionMarker: true, + }).pipe( + Stream.runForEach((item) => + Queue.offer(shellItems, item).pipe(Effect.asVoid), + ), + Effect.forkScoped, + ); + const initialShellItems = yield* collectQueueUntil( + shellItems, + (item) => item.kind === "synchronized", + ); + assert.isFalse(initialShellItems.some((item) => item.kind === "snapshot")); - const threadSnapshot = yield* measureHttpGet({ - url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`, - headers: { cookie }, - }); - assert.equal(threadSnapshot.status, 200); - assert.equal(threadSnapshot.contentEncoding, "gzip"); - const decodedThread = yield* decodeTransferThreadSnapshot( - Buffer.from(threadSnapshot.decodedBody).toString("utf8"), - ); - assert.equal(decodedThread.thread.messages.length, 12); - - const threadItems = yield* Queue.unbounded(); - yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({ - threadId: TRANSFER_THREAD_ID, - afterSequence: decodedThread.snapshotSequence, - requestCompletionMarker: true, - }).pipe( - Stream.runForEach((item) => Queue.offer(threadItems, item).pipe(Effect.asVoid)), - Effect.forkScoped, - ); - const initialThreadItems = yield* collectQueueUntil( - threadItems, - (item) => item.kind === "synchronized", - ); - assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot")); - assert.include(recorder.negotiatedExtensions(), "permessage-deflate"); - - const coldOpenWebSocket = recorder.totals(); - yield* queueMeasuredTransferTurn(harness, provider); - const turnStartTotals = recorder.totals(); - yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ - type: "thread.turn.start", - commandId: CommandId.make(`transfer:${provider}:measured-turn`), - threadId: TRANSFER_THREAD_ID, - message: { - messageId: MessageId.make("transfer-user-measured"), - role: "user", - text: "Measure the client-bound transfer for this turn.", - attachments: [], - }, - modelSelection: transferModelSelection(provider), - runtimeMode: "approval-required", - interactionMode: "default", - createdAt: "2026-06-01T00:06:00.000Z", - }); - yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1); - const finalSequence = yield* harness.engine.latestSequence; + const threadSnapshot = yield* measureHttpGet({ + url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`, + headers: { cookie }, + }); + assert.equal(threadSnapshot.status, 200); + assert.equal(threadSnapshot.contentEncoding, "gzip"); + const decodedThread = yield* decodeTransferThreadSnapshot( + Buffer.from(threadSnapshot.decodedBody).toString("utf8"), + ); + assert.equal( + decodedThread.thread.messages.length, + TRANSFER_HISTORY_TURN_COUNT * 2, + ); - yield* collectQueueUntil( - threadItems, - (item) => item.kind === "event" && item.event.sequence >= finalSequence, - ); - yield* collectQueueUntil( - shellItems, - (item) => - item.kind !== "snapshot" && - item.kind !== "synchronized" && - item.sequence >= finalSequence, - ); - const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals()); + const threadItems = yield* Queue.unbounded(); + yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: TRANSFER_THREAD_ID, + afterSequence: decodedThread.snapshotSequence, + requestCompletionMarker: true, + }).pipe( + Stream.runForEach((item) => + Queue.offer(threadItems, item).pipe(Effect.asVoid), + ), + Effect.forkScoped, + ); + const initialThreadItems = yield* collectQueueUntil( + threadItems, + (item) => item.kind === "synchronized", + ); + assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot")); + assert.include(recorder.negotiatedExtensions(), "permessage-deflate"); + + const coldOpenWebSocket = recorder.totals(); + yield* queueMeasuredTransferTurn(harness, provider); + const turnStartTotals = recorder.totals(); + yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make(`transfer:${provider}:measured-turn`), + threadId: TRANSFER_THREAD_ID, + message: { + messageId: MessageId.make("transfer-user-measured"), + role: "user", + text: "Measure the client-bound transfer for this turn.", + attachments: [], + }, + modelSelection: transferModelSelection(provider), + runtimeMode: "approval-required", + interactionMode: "default", + createdAt: TRANSFER_MEASURED_TURN_CREATED_AT, + }); + yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1); + const finalSequence = yield* harness.engine.latestSequence; - const finalThreadSnapshot = yield* harness.snapshotQuery - .getThreadDetailSnapshot(TRANSFER_THREAD_ID) - .pipe(Effect.map(Option.getOrThrow)); - const finalAssistant = finalThreadSnapshot.thread.messages.findLast( - (message) => message.role === "assistant", - ); - assert.equal(finalAssistant?.text, expectedMeasuredAssistantText(provider)); - assert.equal(finalAssistant?.streaming, false); - assert.equal(finalThreadSnapshot.thread.session?.status, "ready"); - assert.equal(finalThreadSnapshot.thread.checkpoints.length, 7); - - return { - provider, - document, - shellSnapshot, - threadSnapshot, - coldOpenWebSocket, - measuredTurnWebSocket, - } satisfies TransferBudgetRun; - }).pipe(Effect.provide(protocolLayer)), - ); - }), - (harness) => harness.dispose, - ).pipe(Effect.provide(NodeHttpServerTestWithWsDeflate)), - { concurrency: 1 }, - ); + yield* collectQueueUntil( + threadItems, + (item) => item.kind === "event" && item.event.sequence >= finalSequence, + ); + yield* collectQueueUntil( + shellItems, + (item) => + item.kind !== "snapshot" && + item.kind !== "synchronized" && + item.sequence >= finalSequence, + ); + const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals()); + + const finalThreadSnapshot = yield* harness.snapshotQuery + .getThreadDetailSnapshot(TRANSFER_THREAD_ID) + .pipe(Effect.map(Option.getOrThrow)); + const expectedAssistantText = expectedMeasuredAssistantText(provider); + const measuredAssistant = finalThreadSnapshot.thread.messages.find( + (message) => + message.role === "assistant" && message.text === expectedAssistantText, + ); + assert.isDefined(measuredAssistant); + assert.isTrue( + finalThreadSnapshot.thread.messages.length >= TRANSFER_HISTORY_TURN_COUNT * 2, + ); + assert.equal(measuredAssistant?.streaming, false); + assert.equal(finalThreadSnapshot.thread.session?.status, "ready"); + assert.equal( + finalThreadSnapshot.thread.checkpoints.length, + TRANSFER_HISTORY_TURN_COUNT + 1, + ); - const report = formatTransferBudgetReport(runs); - yield* Effect.logInfo(`\n${report}`); - const reportPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_REPORT_PATH").pipe( - Config.option, - ); - if (Option.isSome(reportPath)) { - const fileSystem = yield* FileSystem.FileSystem; - yield* fileSystem.writeFileString(reportPath.value, report); - } - assert.deepEqual(transferBudgetViolations(runs), []); - }).pipe(Effect.provide(NodeServices.layer)), + return { + provider, + document, + shellSnapshot, + threadSnapshot, + coldOpenWebSocket, + measuredTurnWebSocket, + } satisfies TransferBudgetRun; + }).pipe(Effect.provide(protocolLayer)), + ); + }), + (harness) => harness.dispose, + ).pipe(Effect.provide(NodeHttpServerTestWithWsDeflate)), + { concurrency: 1 }, + ); + + const report = formatTransferBudgetReport(runs); + yield* Effect.logInfo(`\n${report}`); + const reportPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_REPORT_PATH").pipe( + Config.option, + ); + if (Option.isSome(reportPath)) { + const fileSystem = yield* FileSystem.FileSystem; + yield* fileSystem.writeFileString(reportPath.value, report); + } + assert.deepEqual(transferBudgetViolations(runs), []); + }).pipe(Effect.provide(NodeServices.layer)), + 120_000, ); From 797fef85226d14702b9a92a3446a2dcd04371c3b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 14:24:58 -0700 Subject: [PATCH 03/10] test(server): keep transfer stress test fast --- .../TransferBudgetReport.integration.ts | 12 +++++----- .../integration/fixtures/transferBudget.ts | 22 +++++-------------- 2 files changed, 12 insertions(+), 22 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 4b3251b713a..74852bcd4d0 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -42,9 +42,9 @@ export const TRANSFER_BUDGETS: Readonly> shellSnapshotWireBytes: 1_000, threadSnapshotWireBytes: 2_550_000, coldOpenWebSocketWireBytes: 1_600, - measuredTurnWebSocketWireBytes: 360_000, - measuredTurnWebSocketDecodedBytes: 2_050_000, - measuredTurnWebSocketMessages: 550, + measuredTurnWebSocketWireBytes: 315_000, + measuredTurnWebSocketDecodedBytes: 1_550_000, + measuredTurnWebSocketMessages: 32, }, claudeAgent: { totalWireBytes: 2_900_000, @@ -52,9 +52,9 @@ export const TRANSFER_BUDGETS: Readonly> shellSnapshotWireBytes: 1_000, threadSnapshotWireBytes: 2_550_000, coldOpenWebSocketWireBytes: 1_600, - measuredTurnWebSocketWireBytes: 360_000, - measuredTurnWebSocketDecodedBytes: 2_050_000, - measuredTurnWebSocketMessages: 550, + measuredTurnWebSocketWireBytes: 315_000, + measuredTurnWebSocketDecodedBytes: 1_550_000, + measuredTurnWebSocketMessages: 32, }, }; diff --git a/apps/server/integration/fixtures/transferBudget.ts b/apps/server/integration/fixtures/transferBudget.ts index dec986a2590..9960e9b2043 100644 --- a/apps/server/integration/fixtures/transferBudget.ts +++ b/apps/server/integration/fixtures/transferBudget.ts @@ -9,8 +9,8 @@ const FIXTURE_THREAD_ID = "transfer-budget-thread"; const FIXTURE_TURN_ID = "transfer-budget-turn"; export const TRANSFER_HISTORY_TURN_COUNT = 10; -export const TRANSFER_HISTORY_TOOLS_PER_TURN = 100; -export const TRANSFER_MEASURED_TOOLS = 150; +export const TRANSFER_HISTORY_TOOLS_PER_TURN = 5; +export const TRANSFER_MEASURED_TOOLS = 20; export const TRANSFER_HISTORY_MCP_RESULT_BYTES = 900_000; export const TRANSFER_MEASURED_MCP_RESULT_BYTES = 1_100_000; @@ -92,16 +92,6 @@ function diagnosticOutput(input: { return chunks.join("").slice(0, input.targetBytes); } -function toolOutputBytes(toolIndex: number, measuredTurn: boolean): number { - if (measuredTurn) { - if (toolIndex >= 136) return 35_000; - if (toolIndex >= 101) return 8_000; - return 1_000; - } - - return 1_000; -} - function assistantChunks(provider: ProviderDriverKind, turnIndex: number): ReadonlyArray { const providerName = provider === "codex" ? "Codex" : "Claude"; const paragraphs: string[] = [ @@ -159,9 +149,9 @@ function baseEvent( /** * Synthetic canonical events calibrated from heavy local Codex and Claude - * threads. Ten historical turns produce roughly 2,000 command activity events - * plus 9 MB of retained MCP results without committing user content. Command - * output is intentionally modest because the client projection strips it. + * threads. Ten historical turns produce 9 MB of retained MCP results without + * committing user content. Command output is intentionally modest because the + * client projection strips it. */ export function makeRecordedTransferTurn( provider: ProviderDriverKind, @@ -243,7 +233,7 @@ export function makeRecordedTransferTurn( provider, turnIndex, toolIndex, - targetBytes: toolOutputBytes(toolIndex, measuredTurn), + targetBytes: 1_000, }), exitCode: 0, durationMs: 500 + toolIndex, From db63269adbd8550589e7ec26faf7481692e849a3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 14:52:58 -0700 Subject: [PATCH 04/10] test(server): scope transfer budget to thread data --- .../TransferBudgetReport.integration.ts | 76 ++++------------- .../integration/fixtures/transferBudget.ts | 3 - apps/server/src/server.test.ts | 81 ++++--------------- 3 files changed, 33 insertions(+), 127 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 74852bcd4d0..6ed829278a1 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -14,19 +14,13 @@ import { export interface TransferBudgetRun { readonly provider: ProviderDriverKind; - readonly document: HttpTransferMeasurement; - readonly shellSnapshot: HttpTransferMeasurement; readonly threadSnapshot: HttpTransferMeasurement; - readonly coldOpenWebSocket: WebSocketTransferTotals; readonly measuredTurnWebSocket: WebSocketTransferTotals; } interface ProviderTransferBudget { readonly totalWireBytes: number; - readonly documentWireBytes: number; - readonly shellSnapshotWireBytes: number; readonly threadSnapshotWireBytes: number; - readonly coldOpenWebSocketWireBytes: number; readonly measuredTurnWebSocketWireBytes: number; readonly measuredTurnWebSocketDecodedBytes: number; readonly measuredTurnWebSocketMessages: number; @@ -35,37 +29,21 @@ interface ProviderTransferBudget { // These caps leave roughly 30% headroom above the deterministic fixture. The // CI report preserves the exact values so intentional protocol growth can be // reviewed and the caps raised explicitly. +const TRANSFER_BUDGET = { + totalWireBytes: 2_900_000, + threadSnapshotWireBytes: 2_600_000, + measuredTurnWebSocketWireBytes: 320_000, + measuredTurnWebSocketDecodedBytes: 1_550_000, + measuredTurnWebSocketMessages: 20, +} satisfies ProviderTransferBudget; + export const TRANSFER_BUDGETS: Readonly> = { - codex: { - totalWireBytes: 2_900_000, - documentWireBytes: 1_700, - shellSnapshotWireBytes: 1_000, - threadSnapshotWireBytes: 2_550_000, - coldOpenWebSocketWireBytes: 1_600, - measuredTurnWebSocketWireBytes: 315_000, - measuredTurnWebSocketDecodedBytes: 1_550_000, - measuredTurnWebSocketMessages: 32, - }, - claudeAgent: { - totalWireBytes: 2_900_000, - documentWireBytes: 1_700, - shellSnapshotWireBytes: 1_000, - threadSnapshotWireBytes: 2_550_000, - coldOpenWebSocketWireBytes: 1_600, - measuredTurnWebSocketWireBytes: 315_000, - measuredTurnWebSocketDecodedBytes: 1_550_000, - measuredTurnWebSocketMessages: 32, - }, + codex: TRANSFER_BUDGET, + claudeAgent: TRANSFER_BUDGET, }; function totalWireBytes(run: TransferBudgetRun): number { - return ( - run.document.wireBytes + - run.shellSnapshot.wireBytes + - run.threadSnapshot.wireBytes + - run.coldOpenWebSocket.wireBytes + - run.measuredTurnWebSocket.wireBytes - ); + return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes; } function formatBytes(bytes: number): string { @@ -97,15 +75,8 @@ export function transferBudgetViolations(runs: ReadonlyArray) continue; } const checks = [ - ["total client-bound wire bytes", totalWireBytes(run), budget.totalWireBytes], - ["document wire bytes", run.document.wireBytes, budget.documentWireBytes], - ["shell snapshot wire bytes", run.shellSnapshot.wireBytes, budget.shellSnapshotWireBytes], + ["total thread wire bytes", totalWireBytes(run), budget.totalWireBytes], ["thread snapshot wire bytes", run.threadSnapshot.wireBytes, budget.threadSnapshotWireBytes], - [ - "cold-open WebSocket wire bytes", - run.coldOpenWebSocket.wireBytes, - budget.coldOpenWebSocketWireBytes, - ], [ "measured-turn WebSocket wire bytes", run.measuredTurnWebSocket.wireBytes, @@ -133,12 +104,12 @@ export function transferBudgetViolations(runs: ReadonlyArray) export function formatTransferBudgetReport(runs: ReadonlyArray): string { const lines = [ - "# T3 Code stress transfer budget", + "# T3 Code thread transfer budget", "", - "Wire values are client-bound bytes read from local HTTP and WebSocket sockets. They include HTTP response headers and the WebSocket upgrade, but exclude TCP/IP and TLS framing. WebSocket permessage-deflate is negotiated.", + "Wire values are thread data bytes read from local HTTP and WebSocket sockets. HTTP includes response headers; WebSocket measurement starts after the resumed thread subscription synchronizes. TCP/IP, TLS framing, and the WebSocket upgrade are excluded. WebSocket permessage-deflate is negotiated.", `Scenario: ${TRANSFER_HISTORY_TURN_COUNT} historical turns with ${TRANSFER_HISTORY_TOOLS_PER_TURN} command tools and one retained ${formatBytes(TRANSFER_HISTORY_MCP_RESULT_BYTES)} MCP result each, followed by one measured turn with ${TRANSFER_MEASURED_TOOLS} command tools and a retained ${formatBytes(TRANSFER_MEASURED_MCP_RESULT_BYTES)} MCP result. Payload sizes are calibrated from heavy local Codex and Claude histories and contain no user data.`, "", - "| Provider | Total client-bound wire | Budget | Result |", + "| Provider | Total thread wire | Budget | Result |", "| --- | ---: | ---: | --- |", ...runs.flatMap((run) => { const budget = TRANSFER_BUDGETS[run.provider]; @@ -159,14 +130,6 @@ export function formatTransferBudgetReport(runs: ReadonlyArray( queue: Queue.Queue, predicate: (value: A) => boolean, + waitDescription: string, ) { - const values: A[] = []; - while (true) { - const value = yield* Queue.take(queue); - values.push(value); - if (predicate(value)) return values; - } + return yield* Effect.gen(function* () { + const values: A[] = []; + while (true) { + const value = yield* Queue.take(queue); + values.push(value); + if (predicate(value)) return values; + } + }).pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => Effect.die(new Error(`Timed out waiting for ${waitDescription}`)), + }), + ); }); import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; @@ -7895,14 +7897,13 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }); it.live( - "reports stress HTTP and WebSocket transfer budgets", + "reports thread HTTP and WebSocket transfer budgets", () => Effect.gen(function* () { const providers = [ ProviderDriverKind.make("codex"), ProviderDriverKind.make("claudeAgent"), ] as const; - const staticDir = NodeURL.fileURLToPath(new URL("../../web", import.meta.url)); const runs = yield* Effect.forEach( providers, @@ -7913,7 +7914,6 @@ it.live( Effect.gen(function* () { yield* seedTransferBudgetHistory(harness, provider); yield* buildAppUnderTest({ - config: { staticDir }, layers: { orchestrationEngine: harness.engine, projectionSnapshotQuery: harness.snapshotQuery, @@ -7922,8 +7922,6 @@ it.live( const baseUrl = yield* getHttpServerUrl(); const cookie = yield* getAuthenticatedSessionCookieHeader(); - const document = yield* measureHttpGet({ url: `${baseUrl}/` }); - assert.equal(document.status, 200); const recorder = makeWebSocketTransferRecorder(); const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws"; @@ -7936,42 +7934,6 @@ it.live( return yield* Effect.scoped( Effect.gen(function* () { const client = yield* makeCountingWsRpcClient; - yield* client[WS_METHODS.serverGetConfig]({}); - - const configEvents = yield* Queue.unbounded(); - yield* client[WS_METHODS.subscribeServerConfig]({}).pipe( - Stream.runForEach((event) => - Queue.offer(configEvents, event).pipe(Effect.asVoid), - ), - Effect.forkScoped, - ); - yield* Queue.take(configEvents); - - const shellSnapshot = yield* measureHttpGet({ - url: `${baseUrl}/api/orchestration/shell`, - headers: { cookie }, - }); - assert.equal(shellSnapshot.status, 200); - assert.equal(shellSnapshot.contentEncoding, "gzip"); - const decodedShell = yield* decodeTransferShellSnapshot( - Buffer.from(shellSnapshot.decodedBody).toString("utf8"), - ); - - const shellItems = yield* Queue.unbounded(); - yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({ - afterSequence: decodedShell.snapshotSequence, - requestCompletionMarker: true, - }).pipe( - Stream.runForEach((item) => - Queue.offer(shellItems, item).pipe(Effect.asVoid), - ), - Effect.forkScoped, - ); - const initialShellItems = yield* collectQueueUntil( - shellItems, - (item) => item.kind === "synchronized", - ); - assert.isFalse(initialShellItems.some((item) => item.kind === "snapshot")); const threadSnapshot = yield* measureHttpGet({ url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`, @@ -8001,11 +7963,11 @@ it.live( const initialThreadItems = yield* collectQueueUntil( threadItems, (item) => item.kind === "synchronized", + `${provider} thread subscription to synchronize`, ); assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot")); assert.include(recorder.negotiatedExtensions(), "permessage-deflate"); - const coldOpenWebSocket = recorder.totals(); yield* queueMeasuredTransferTurn(harness, provider); const turnStartTotals = recorder.totals(); yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ @@ -8029,13 +7991,7 @@ it.live( yield* collectQueueUntil( threadItems, (item) => item.kind === "event" && item.event.sequence >= finalSequence, - ); - yield* collectQueueUntil( - shellItems, - (item) => - item.kind !== "snapshot" && - item.kind !== "synchronized" && - item.sequence >= finalSequence, + `${provider} thread stream to reach sequence ${finalSequence}`, ); const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals()); @@ -8060,10 +8016,7 @@ it.live( return { provider, - document, - shellSnapshot, threadSnapshot, - coldOpenWebSocket, measuredTurnWebSocket, } satisfies TransferBudgetRun; }).pipe(Effect.provide(protocolLayer)), From 0da71752e827370e34bc1470abe51a084db47cf1 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 16:33:09 -0700 Subject: [PATCH 05/10] ci: comment thread transfer impact on pull requests --- .github/scripts/thread-transfer-report.cjs | 368 ++++++++++++++++++ .../scripts/thread-transfer-report.test.cjs | 141 +++++++ .github/workflows/ci.yml | 10 + .github/workflows/thread-transfer-report.yml | 75 ++++ .../TransferBudgetReport.integration.ts | 40 +- apps/server/src/server.test.ts | 8 + 6 files changed, 641 insertions(+), 1 deletion(-) create mode 100644 .github/scripts/thread-transfer-report.cjs create mode 100644 .github/scripts/thread-transfer-report.test.cjs create mode 100644 .github/workflows/thread-transfer-report.yml diff --git a/.github/scripts/thread-transfer-report.cjs b/.github/scripts/thread-transfer-report.cjs new file mode 100644 index 00000000000..a7fbb77cdc6 --- /dev/null +++ b/.github/scripts/thread-transfer-report.cjs @@ -0,0 +1,368 @@ +const fs = require("node:fs"); +const path = require("node:path"); + +const ARTIFACT_NAME = "thread-transfer-results"; +const RESULT_FILE = "thread-transfer-result.json"; +const COMMENT_MARKER = ""; +const PROVIDERS = ["codex", "claudeAgent"]; +const OBSERVED_KEYS = [ + "totalWireBytes", + "threadSnapshotWireBytes", + "threadSnapshotDecodedBytes", + "measuredTurnWebSocketWireBytes", + "measuredTurnWebSocketDecodedBytes", + "measuredTurnWebSocketMessages", +]; +const CEILING_KEYS = [ + "totalWireBytes", + "threadSnapshotWireBytes", + "measuredTurnWebSocketWireBytes", + "measuredTurnWebSocketDecodedBytes", + "measuredTurnWebSocketMessages", +]; +const SCENARIO_KEYS = [ + "id", + "historyTurns", + "historyCommandToolsPerTurn", + "historyMcpResultBytes", + "measuredCommandTools", + "measuredMcpResultBytes", +]; + +function assertObject(value, label) { + if (value === null || typeof value !== "object" || Array.isArray(value)) { + throw new Error(`${label} must be an object`); + } +} + +function assertExactKeys(value, expected, label) { + assertObject(value, label); + const actual = Object.keys(value).sort(); + const wanted = [...expected].sort(); + if (actual.length !== wanted.length || actual.some((key, index) => key !== wanted[index])) { + throw new Error(`${label} has unexpected fields`); + } +} + +function assertMetric(value, label) { + if (!Number.isSafeInteger(value) || value < 0 || value > 1_000_000_000) { + throw new Error(`${label} must be a non-negative safe integer below 1,000,000,000`); + } +} + +function validateResult(value) { + assertExactKeys(value, ["schemaVersion", "scenario", "providers"], "result"); + if (value.schemaVersion !== 1) { + throw new Error("result.schemaVersion must be 1"); + } + + assertExactKeys(value.scenario, SCENARIO_KEYS, "result.scenario"); + if (value.scenario.id !== "thread-transfer-v1") { + throw new Error("result.scenario.id is not supported"); + } + for (const key of SCENARIO_KEYS.slice(1)) { + assertMetric(value.scenario[key], `result.scenario.${key}`); + } + + assertExactKeys(value.providers, PROVIDERS, "result.providers"); + for (const provider of PROVIDERS) { + const entry = value.providers[provider]; + assertExactKeys(entry, ["observed", "ceiling"], `result.providers.${provider}`); + assertExactKeys(entry.observed, OBSERVED_KEYS, `result.providers.${provider}.observed`); + assertExactKeys(entry.ceiling, CEILING_KEYS, `result.providers.${provider}.ceiling`); + for (const key of OBSERVED_KEYS) { + assertMetric(entry.observed[key], `result.providers.${provider}.observed.${key}`); + } + for (const key of CEILING_KEYS) { + assertMetric(entry.ceiling[key], `result.providers.${provider}.ceiling.${key}`); + } + } + + return value; +} + +function readResult(directory) { + if (!directory) return undefined; + const file = path.join(directory, RESULT_FILE); + if (!fs.existsSync(file)) return undefined; + const stat = fs.lstatSync(file); + if (!stat.isFile() || stat.size > 64 * 1_024) { + throw new Error("thread transfer result must be a regular file smaller than 64 KiB"); + } + return validateResult(JSON.parse(fs.readFileSync(file, "utf8"))); +} + +function formatBytes(bytes) { + if (bytes < 1_024) return `${bytes} B`; + if (bytes >= 1_024 * 1_024) return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB`; + return `${(bytes / 1_024).toFixed(1)} KiB`; +} + +function formatValue(value, kind) { + return kind === "messages" ? value.toLocaleString("en-US") : formatBytes(value); +} + +function formatImpact(current, baseline, kind) { + if (baseline === undefined) return "—"; + const delta = current - baseline; + const prefix = delta > 0 ? "+" : delta < 0 ? "−" : ""; + const magnitude = formatValue(Math.abs(delta), kind); + const percent = + baseline === 0 ? "" : ` (${prefix}${Math.abs((delta / baseline) * 100).toFixed(1)}%)`; + return `${prefix}${magnitude}${percent}`; +} + +function sameScenario(left, right) { + return SCENARIO_KEYS.every((key) => left[key] === right[key]); +} + +const METRICS = [ + { key: "totalWireBytes", label: "Total thread wire", kind: "bytes" }, + { key: "threadSnapshotWireBytes", label: "Thread snapshot wire", kind: "bytes" }, + { + key: "measuredTurnWebSocketWireBytes", + label: "Live turn WebSocket wire", + kind: "bytes", + }, + { + key: "measuredTurnWebSocketDecodedBytes", + label: "Live turn WebSocket decoded", + kind: "bytes", + }, + { key: "measuredTurnWebSocketMessages", label: "Live turn messages", kind: "messages" }, +]; + +function renderComment(input) { + const current = input.current; + const baseline = input.baseline; + const comparable = baseline !== undefined && sameScenario(current.scenario, baseline.scenario); + const rows = []; + const ceilingChanges = []; + let failed = false; + + for (const provider of PROVIDERS) { + for (const metric of METRICS) { + const observed = current.providers[provider].observed[metric.key]; + const ceiling = current.providers[provider].ceiling[metric.key]; + const baselineObserved = comparable + ? baseline.providers[provider].observed[metric.key] + : undefined; + const pass = observed <= ceiling; + failed ||= !pass; + rows.push( + `| ${provider === "codex" ? "Codex" : "Claude"} | ${metric.label} | ${baselineObserved === undefined ? "—" : formatValue(baselineObserved, metric.kind)} | ${formatValue(observed, metric.kind)} | ${formatImpact(observed, baselineObserved, metric.kind)} | ${formatValue(ceiling, metric.kind)} | ${pass ? "✅" : "❌"} |`, + ); + + if (baseline && baseline.providers[provider].ceiling[metric.key] !== ceiling) { + ceilingChanges.push( + `- ${provider === "codex" ? "Codex" : "Claude"} ${metric.label}: ${formatValue(baseline.providers[provider].ceiling[metric.key], metric.kind)} → ${formatValue(ceiling, metric.kind)}`, + ); + } + } + } + + const baselineLink = input.baselineRun + ? `[\`${input.baselineRun.sha.slice(0, 7)}\`](${input.baselineRun.url})` + : "unavailable"; + const currentLink = `[\`${input.currentRun.sha.slice(0, 7)}\`](${input.currentRun.url})`; + const notices = []; + if (!baseline) { + notices.push( + "> ℹ️ No successful `main` baseline artifact is available yet. This run establishes the initial measurement.", + ); + } else if (!comparable) { + notices.push( + "> ⚠️ The thread fixture changed, so impact percentages are not directly comparable to the `main` baseline.", + ); + } else if (!input.baselineRun.matchesBase) { + notices.push( + "> ℹ️ The exact PR base did not have a successful artifact. Baseline uses the latest successful `main` measurement shown below.", + ); + } + if (ceilingChanges.length > 0) { + notices.push( + `> ⚠️ **This PR changes transfer ceilings:**\n>\n${ceilingChanges.map((line) => `> ${line}`).join("\n")}`, + ); + } + + return [ + COMMENT_MARKER, + "## Thread transfer impact", + "", + failed + ? "❌ One or more thread transfer ceilings were exceeded." + : "✅ Thread transfer remains within every enforced ceiling.", + ...(notices.length > 0 ? ["", ...notices] : []), + "", + "| Provider | Metric | Main baseline | This PR | Impact | PR ceiling | |", + "| --- | --- | ---: | ---: | ---: | ---: | --- |", + ...rows, + "", + `Baseline: ${baselineLink} · PR result: ${currentLink} · Source CI: ${input.currentRun.conclusion}`, + "", + "
", + "Scenario and decoded snapshot size", + "", + `${current.scenario.historyTurns} historical turns, ${current.scenario.historyCommandToolsPerTurn} command tools per turn, ${formatBytes(current.scenario.historyMcpResultBytes)} retained MCP result per historical turn, and a ${formatBytes(current.scenario.measuredMcpResultBytes)} retained result in the measured turn.`, + "", + ...PROVIDERS.map( + (provider) => + `- ${provider === "codex" ? "Codex" : "Claude"} decoded thread snapshot: ${formatBytes(current.providers[provider].observed.threadSnapshotDecodedBytes)}`, + ), + "", + "
", + "", + "_Updated in place by a trusted workflow. PR artifacts are strictly validated and never executed._", + ].join("\n"); +} + +async function artifactsForRun(github, owner, repo, runId) { + return github.paginate(github.rest.actions.listWorkflowRunArtifacts, { + owner, + repo, + run_id: runId, + per_page: 100, + }); +} + +function findResultArtifact(artifacts) { + return artifacts.find((artifact) => artifact.name === ARTIFACT_NAME && !artifact.expired); +} + +async function resolve({ github, context, core }) { + const source = context.payload.workflow_run; + const { owner, repo } = context.repo; + if (source.event !== "pull_request") { + core.setOutput("publish", "false"); + return; + } + + let pullNumber = source.pull_requests?.[0]?.number; + if (!pullNumber) { + const associated = await github.paginate( + github.rest.repos.listPullRequestsAssociatedWithCommit, + { owner, repo, commit_sha: source.head_sha, per_page: 100 }, + ); + pullNumber = associated.find((pull) => pull.state === "open")?.number; + } + if (!pullNumber) { + core.info("No open pull request is associated with the completed CI run."); + core.setOutput("publish", "false"); + return; + } + + const { data: pull } = await github.rest.pulls.get({ owner, repo, pull_number: pullNumber }); + if (pull.head.sha !== source.head_sha) { + core.info(`Skipping stale CI result ${source.head_sha}; PR head is ${pull.head.sha}.`); + core.setOutput("publish", "false"); + return; + } + + const sourceArtifacts = await artifactsForRun(github, owner, repo, source.id); + const sourceArtifact = findResultArtifact(sourceArtifacts); + const workflowRuns = await github.paginate(github.rest.actions.listWorkflowRuns, { + owner, + repo, + workflow_id: source.workflow_id, + branch: pull.base.ref, + event: "push", + status: "success", + per_page: 100, + }); + const orderedRuns = [ + ...workflowRuns.filter((run) => run.head_sha === pull.base.sha), + ...workflowRuns.filter((run) => run.head_sha !== pull.base.sha), + ].slice(0, 20); + + let baselineRun; + for (const run of orderedRuns) { + const artifacts = await artifactsForRun(github, owner, repo, run.id); + if (findResultArtifact(artifacts)) { + baselineRun = run; + break; + } + } + + core.setOutput("publish", "true"); + core.setOutput("pull_number", String(pullNumber)); + core.setOutput("pr_artifact", sourceArtifact ? "true" : "false"); + core.setOutput("pr_run_id", String(source.id)); + core.setOutput("pr_sha", source.head_sha); + core.setOutput("pr_conclusion", source.conclusion ?? "unknown"); + core.setOutput("baseline_artifact", baselineRun ? "true" : "false"); + core.setOutput("baseline_run_id", baselineRun ? String(baselineRun.id) : ""); + core.setOutput("baseline_sha", baselineRun?.head_sha ?? ""); + core.setOutput( + "baseline_matches_base", + baselineRun?.head_sha === pull.base.sha ? "true" : "false", + ); +} + +async function upsertComment(github, context, pullNumber, body) { + const { owner, repo } = context.repo; + const comments = await github.paginate(github.rest.issues.listComments, { + owner, + repo, + issue_number: pullNumber, + per_page: 100, + }); + const existing = comments.find( + (comment) => + comment.user?.login === "github-actions[bot]" && comment.body?.includes(COMMENT_MARKER), + ); + if (existing) { + await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body }); + } else { + await github.rest.issues.createComment({ owner, repo, issue_number: pullNumber, body }); + } +} + +async function publish({ github, context, core }) { + const pullNumber = Number(process.env.PR_NUMBER); + if (!Number.isSafeInteger(pullNumber) || pullNumber <= 0) { + throw new Error("PR_NUMBER is invalid"); + } + + const current = readResult(process.env.PR_RESULT_DIR); + const currentRun = { + sha: process.env.PR_SHA, + conclusion: process.env.PR_CONCLUSION, + url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.PR_RUN_ID}`, + }; + if (!current) { + await upsertComment( + github, + context, + pullNumber, + [ + COMMENT_MARKER, + "## Thread transfer impact", + "", + `⚠️ The latest [CI run](${currentRun.url}) did not produce a thread transfer result for \`${currentRun.sha.slice(0, 7)}\`.`, + "", + "_This comment will update automatically after the next completed run._", + ].join("\n"), + ); + return; + } + + const baseline = readResult(process.env.BASELINE_RESULT_DIR); + const baselineRun = baseline + ? { + sha: process.env.BASELINE_SHA, + matchesBase: process.env.BASELINE_MATCHES_BASE === "true", + url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.BASELINE_RUN_ID}`, + } + : undefined; + const body = renderComment({ current, baseline, currentRun, baselineRun }); + await upsertComment(github, context, pullNumber, body); + core.info(`Updated thread transfer report on PR #${pullNumber}.`); +} + +module.exports = { + publish, + readResult, + renderComment, + resolve, + validateResult, +}; diff --git a/.github/scripts/thread-transfer-report.test.cjs b/.github/scripts/thread-transfer-report.test.cjs new file mode 100644 index 00000000000..f85d1ea0e5b --- /dev/null +++ b/.github/scripts/thread-transfer-report.test.cjs @@ -0,0 +1,141 @@ +const assert = require("node:assert/strict"); +const test = require("node:test"); + +const { renderComment, resolve, validateResult } = require("./thread-transfer-report.cjs"); + +function result(overrides = {}) { + const observed = { + totalWireBytes: 2_200_000, + threadSnapshotWireBytes: 1_950_000, + threadSnapshotDecodedBytes: 9_100_000, + measuredTurnWebSocketWireBytes: 250_000, + measuredTurnWebSocketDecodedBytes: 1_150_000, + measuredTurnWebSocketMessages: 15, + }; + const ceiling = { + totalWireBytes: 2_900_000, + threadSnapshotWireBytes: 2_600_000, + measuredTurnWebSocketWireBytes: 320_000, + measuredTurnWebSocketDecodedBytes: 1_550_000, + measuredTurnWebSocketMessages: 20, + }; + return { + schemaVersion: 1, + scenario: { + id: "thread-transfer-v1", + historyTurns: 10, + historyCommandToolsPerTurn: 5, + historyMcpResultBytes: 900_000, + measuredCommandTools: 20, + measuredMcpResultBytes: 1_100_000, + }, + providers: { + codex: { observed: { ...observed, ...overrides }, ceiling }, + claudeAgent: { observed, ceiling }, + }, + }; +} + +test("validates the fixed artifact schema", () => { + assert.equal(validateResult(result()).schemaVersion, 1); + assert.throws( + () => validateResult({ ...result(), injectedMarkdown: "@everyone" }), + /unexpected fields/, + ); + assert.throws( + () => validateResult(result({ totalWireBytes: "lots" })), + /non-negative safe integer/, + ); +}); + +test("renders baseline, impact, ceiling, and ceiling changes", () => { + const baseline = result(); + const current = result({ measuredTurnWebSocketWireBytes: 260_000 }); + current.providers.codex.ceiling = { + ...current.providers.codex.ceiling, + measuredTurnWebSocketWireBytes: 330_000, + }; + const comment = renderComment({ + current, + baseline, + currentRun: { + sha: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + conclusion: "success", + url: "https://github.com/pingdotgg/t3code/actions/runs/2", + }, + baselineRun: { + sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + matchesBase: true, + url: "https://github.com/pingdotgg/t3code/actions/runs/1", + }, + }); + + assert.match(comment, /Main baseline \| This PR \| Impact \| PR ceiling/); + assert.match(comment, /\+9\.8 KiB \(\+4\.0%\)/); + assert.match(comment, /This PR changes transfer ceilings/); + assert.match(comment, /312\.5 KiB → 322\.3 KiB/); + assert.match(comment, //); +}); + +test("resolves the current PR artifact and exact main baseline", async () => { + const outputs = {}; + const listWorkflowRunArtifacts = () => {}; + const listWorkflowRuns = () => {}; + const github = { + paginate: async (method, input) => { + if (method === listWorkflowRunArtifacts) { + return [ + { + name: "thread-transfer-results", + expired: false, + runId: input.run_id, + }, + ]; + } + if (method === listWorkflowRuns) { + return [{ id: 1, head_sha: "base-sha" }]; + } + throw new Error("unexpected pagination call"); + }, + rest: { + actions: { listWorkflowRunArtifacts, listWorkflowRuns }, + pulls: { + get: async () => ({ + data: { + head: { sha: "head-sha" }, + base: { sha: "base-sha", ref: "main" }, + }, + }), + }, + repos: { listPullRequestsAssociatedWithCommit: () => {} }, + }, + }; + await resolve({ + github, + context: { + repo: { owner: "pingdotgg", repo: "t3code" }, + payload: { + workflow_run: { + id: 2, + event: "pull_request", + workflow_id: 3, + head_sha: "head-sha", + conclusion: "success", + pull_requests: [{ number: 5350 }], + }, + }, + }, + core: { + info: () => {}, + setOutput: (key, value) => { + outputs[key] = value; + }, + }, + }); + + assert.equal(outputs.publish, "true"); + assert.equal(outputs.pull_number, "5350"); + assert.equal(outputs.pr_artifact, "true"); + assert.equal(outputs.baseline_run_id, "1"); + assert.equal(outputs.baseline_matches_base, "true"); +}); diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4845f63a1e8..052a8c20cf7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -86,6 +86,7 @@ jobs: - name: Test env: T3CODE_TRANSFER_BUDGET_REPORT_PATH: ${{ runner.temp }}/t3code-transfer-budget.md + T3CODE_TRANSFER_BUDGET_RESULT_PATH: ${{ runner.temp }}/thread-transfer-result.json run: vp run test - name: Publish transfer budget report @@ -97,6 +98,15 @@ jobs: echo "Transfer budget report was not produced." >> "$GITHUB_STEP_SUMMARY" fi + - name: Upload thread transfer result + if: always() + uses: actions/upload-artifact@v7 + with: + name: thread-transfer-results + path: ${{ runner.temp }}/thread-transfer-result.json + if-no-files-found: ignore + retention-days: 30 + - name: Test resource monitor run: cargo test --locked --manifest-path native/resource-monitor/Cargo.toml diff --git a/.github/workflows/thread-transfer-report.yml b/.github/workflows/thread-transfer-report.yml new file mode 100644 index 00000000000..23eec72923b --- /dev/null +++ b/.github/workflows/thread-transfer-report.yml @@ -0,0 +1,75 @@ +name: Thread Transfer Report + +on: + workflow_run: + workflows: [CI] + types: [completed] + +permissions: + actions: read + contents: read + pull-requests: write + +jobs: + publish: + name: Publish PR comment + if: github.event.workflow_run.event == 'pull_request' + runs-on: ubuntu-24.04 + concurrency: + group: thread-transfer-report-${{ github.event.workflow_run.pull_requests[0].number || github.event.workflow_run.id }} + cancel-in-progress: true + steps: + # workflow_run has a write-capable token even for fork PRs. Only load the + # publisher from the trusted default branch and never execute PR code. + - name: Checkout trusted publisher + uses: actions/checkout@v6 + with: + ref: ${{ github.event.repository.default_branch }} + sparse-checkout: .github/scripts + + - name: Test trusted publisher + run: node --test .github/scripts/thread-transfer-report.test.cjs + + - id: resolve + name: Resolve PR and baseline artifacts + uses: actions/github-script@v8 + with: + script: | + const reporter = require("./.github/scripts/thread-transfer-report.cjs"); + await reporter.resolve({ github, context, core }); + + - name: Download PR result + if: steps.resolve.outputs.publish == 'true' && steps.resolve.outputs.pr_artifact == 'true' + uses: actions/download-artifact@v8 + with: + name: thread-transfer-results + path: ${{ runner.temp }}/thread-transfer/pr + github-token: ${{ secrets.GITHUB_TOKEN }} + run-id: ${{ steps.resolve.outputs.pr_run_id }} + + - name: Download main baseline + if: steps.resolve.outputs.publish == 'true' && steps.resolve.outputs.baseline_artifact == 'true' + uses: actions/download-artifact@v8 + with: + name: thread-transfer-results + path: ${{ runner.temp }}/thread-transfer/main + github-token: ${{ secrets.GITHUB_TOKEN }} + run-id: ${{ steps.resolve.outputs.baseline_run_id }} + + - name: Update thread transfer comment + if: steps.resolve.outputs.publish == 'true' + uses: actions/github-script@v8 + env: + PR_NUMBER: ${{ steps.resolve.outputs.pull_number }} + PR_SHA: ${{ steps.resolve.outputs.pr_sha }} + PR_CONCLUSION: ${{ steps.resolve.outputs.pr_conclusion }} + PR_RUN_ID: ${{ steps.resolve.outputs.pr_run_id }} + PR_RESULT_DIR: ${{ runner.temp }}/thread-transfer/pr + BASELINE_SHA: ${{ steps.resolve.outputs.baseline_sha }} + BASELINE_MATCHES_BASE: ${{ steps.resolve.outputs.baseline_matches_base }} + BASELINE_RUN_ID: ${{ steps.resolve.outputs.baseline_run_id }} + BASELINE_RESULT_DIR: ${{ runner.temp }}/thread-transfer/main + with: + script: | + const reporter = require("./.github/scripts/thread-transfer-report.cjs"); + await reporter.publish({ github, context, core }); diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 6ed829278a1..ba5166f48c1 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -46,6 +46,44 @@ function totalWireBytes(run: TransferBudgetRun): number { return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes; } +function observedTransfer(run: TransferBudgetRun) { + return { + totalWireBytes: totalWireBytes(run), + threadSnapshotWireBytes: run.threadSnapshot.wireBytes, + threadSnapshotDecodedBytes: run.threadSnapshot.decodedBodyBytes, + measuredTurnWebSocketWireBytes: run.measuredTurnWebSocket.wireBytes, + measuredTurnWebSocketDecodedBytes: run.measuredTurnWebSocket.decodedBytes, + measuredTurnWebSocketMessages: run.measuredTurnWebSocket.messages, + }; +} + +/** Machine-readable input for the trusted PR comment publisher. */ +export function formatTransferBudgetResult(runs: ReadonlyArray): string { + const providers = Object.fromEntries( + runs.flatMap((run) => { + const ceiling = TRANSFER_BUDGETS[run.provider]; + return ceiling ? [[run.provider, { observed: observedTransfer(run), ceiling }]] : []; + }), + ); + + return `${JSON.stringify( + { + schemaVersion: 1, + scenario: { + id: "thread-transfer-v1", + historyTurns: TRANSFER_HISTORY_TURN_COUNT, + historyCommandToolsPerTurn: TRANSFER_HISTORY_TOOLS_PER_TURN, + historyMcpResultBytes: TRANSFER_HISTORY_MCP_RESULT_BYTES, + measuredCommandTools: TRANSFER_MEASURED_TOOLS, + measuredMcpResultBytes: TRANSFER_MEASURED_MCP_RESULT_BYTES, + }, + providers, + }, + null, + 2, + )}\n`; +} + function formatBytes(bytes: number): string { if (bytes < 1_024) return `${bytes} B`; if (bytes >= 1_024 * 1_024) { @@ -114,7 +152,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray { const budget = TRANSFER_BUDGETS[run.provider]; if (!budget) return []; - const observed = totalWireBytes(run); + const observed = observedTransfer(run).totalWireBytes; return [ `| ${run.provider} | ${formatBytes(observed)} | ${formatBytes(budget.totalWireBytes)} | ${observed <= budget.totalWireBytes ? "PASS" : "FAIL"} |`, ]; diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index ef789f2dfaa..d2c2ee89593 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -172,6 +172,7 @@ import { } from "../integration/TransferBudgetScenario.integration.ts"; import { formatTransferBudgetReport, + formatTransferBudgetResult, type TransferBudgetRun, transferBudgetViolations, } from "../integration/TransferBudgetReport.integration.ts"; @@ -8036,6 +8037,13 @@ it.live( const fileSystem = yield* FileSystem.FileSystem; yield* fileSystem.writeFileString(reportPath.value, report); } + const resultPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_RESULT_PATH").pipe( + Config.option, + ); + if (Option.isSome(resultPath)) { + const fileSystem = yield* FileSystem.FileSystem; + yield* fileSystem.writeFileString(resultPath.value, formatTransferBudgetResult(runs)); + } assert.deepEqual(transferBudgetViolations(runs), []); }).pipe(Effect.provide(NodeServices.layer)), 120_000, From 1490096385b5ebe9c57783ce0959b1418363d3f3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 22:15:31 -0700 Subject: [PATCH 06/10] test(server): wait for complete thread transfer tail --- .github/scripts/thread-transfer-report.cjs | 34 +++++++++++++-- .../scripts/thread-transfer-report.test.cjs | 43 ++++++++++++++++++- .../OrchestrationEngineHarness.integration.ts | 6 +++ .../TransferBudgetScenario.integration.ts | 1 + apps/server/src/server.test.ts | 20 +++++++-- apps/server/src/ws.ts | 2 +- 6 files changed, 97 insertions(+), 9 deletions(-) diff --git a/.github/scripts/thread-transfer-report.cjs b/.github/scripts/thread-transfer-report.cjs index a7fbb77cdc6..da14698ac97 100644 --- a/.github/scripts/thread-transfer-report.cjs +++ b/.github/scripts/thread-transfer-report.cjs @@ -317,6 +317,22 @@ async function upsertComment(github, context, pullNumber, body) { } } +async function upsertCommentForCurrentHead(github, context, core, pullNumber, expectedSha, body) { + const { owner, repo } = context.repo; + const { data: pull } = await github.rest.pulls.get({ + owner, + repo, + pull_number: pullNumber, + }); + if (pull.head.sha !== expectedSha) { + core.info(`Skipping stale CI result ${expectedSha}; PR head is ${pull.head.sha}.`); + return false; + } + + await upsertComment(github, context, pullNumber, body); + return true; +} + async function publish({ github, context, core }) { const pullNumber = Number(process.env.PR_NUMBER); if (!Number.isSafeInteger(pullNumber) || pullNumber <= 0) { @@ -330,10 +346,12 @@ async function publish({ github, context, core }) { url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.PR_RUN_ID}`, }; if (!current) { - await upsertComment( + await upsertCommentForCurrentHead( github, context, + core, pullNumber, + currentRun.sha, [ COMMENT_MARKER, "## Thread transfer impact", @@ -355,8 +373,17 @@ async function publish({ github, context, core }) { } : undefined; const body = renderComment({ current, baseline, currentRun, baselineRun }); - await upsertComment(github, context, pullNumber, body); - core.info(`Updated thread transfer report on PR #${pullNumber}.`); + const published = await upsertCommentForCurrentHead( + github, + context, + core, + pullNumber, + currentRun.sha, + body, + ); + if (published) { + core.info(`Updated thread transfer report on PR #${pullNumber}.`); + } } module.exports = { @@ -364,5 +391,6 @@ module.exports = { readResult, renderComment, resolve, + upsertCommentForCurrentHead, validateResult, }; diff --git a/.github/scripts/thread-transfer-report.test.cjs b/.github/scripts/thread-transfer-report.test.cjs index f85d1ea0e5b..a6a3c40e420 100644 --- a/.github/scripts/thread-transfer-report.test.cjs +++ b/.github/scripts/thread-transfer-report.test.cjs @@ -1,7 +1,12 @@ const assert = require("node:assert/strict"); const test = require("node:test"); -const { renderComment, resolve, validateResult } = require("./thread-transfer-report.cjs"); +const { + renderComment, + resolve, + upsertCommentForCurrentHead, + validateResult, +} = require("./thread-transfer-report.cjs"); function result(overrides = {}) { const observed = { @@ -139,3 +144,39 @@ test("resolves the current PR artifact and exact main baseline", async () => { assert.equal(outputs.baseline_run_id, "1"); assert.equal(outputs.baseline_matches_base, "true"); }); + +test("does not publish a stale result after the PR head advances", async () => { + let listedComments = false; + const info = []; + const published = await upsertCommentForCurrentHead( + { + paginate: async () => { + listedComments = true; + return []; + }, + rest: { + issues: { + listComments: () => {}, + createComment: () => { + throw new Error("must not create a stale comment"); + }, + updateComment: () => { + throw new Error("must not update a stale comment"); + }, + }, + pulls: { + get: async () => ({ data: { head: { sha: "new-head-sha" } } }), + }, + }, + }, + { repo: { owner: "pingdotgg", repo: "t3code" } }, + { info: (message) => info.push(message) }, + 5350, + "old-head-sha", + "stale body", + ); + + assert.equal(published, false); + assert.equal(listedComments, false); + assert.deepEqual(info, ["Skipping stale CI result old-head-sha; PR head is new-head-sha."]); +}); diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index 8daa9555655..d192cbeac8e 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -55,6 +55,7 @@ import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceip import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts"; import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts"; +import { CheckpointReactor } from "../src/orchestration/Services/CheckpointReactor.ts"; import { ProviderRuntimeIngestionService } from "../src/orchestration/Services/ProviderRuntimeIngestion.ts"; import { OrchestrationEngineService, @@ -220,6 +221,7 @@ export interface OrchestrationIntegrationHarness { ): Effect.Effect; }; readonly drainProviderRuntime: Effect.Effect; + readonly drainCheckpointReactor: Effect.Effect; readonly dispose: Effect.Effect; } @@ -398,6 +400,9 @@ export const makeOrchestrationIntegrationHarness = ( "load ProviderRuntimeIngestion service", () => runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)), ).pipe(Effect.orDie); + const checkpointReactor = yield* tryRuntimePromise("load CheckpointReactor service", () => + runtime.runPromise(Effect.service(CheckpointReactor)), + ).pipe(Effect.orDie); const snapshotQuery = yield* tryRuntimePromise("load ProjectionSnapshotQuery service", () => runtime.runPromise(Effect.service(ProjectionSnapshotQuery)), ).pipe(Effect.orDie); @@ -563,6 +568,7 @@ export const makeOrchestrationIntegrationHarness = ( waitForPendingApproval, waitForReceipt, drainProviderRuntime: providerRuntimeIngestion.drain, + drainCheckpointReactor: checkpointReactor.drain, dispose, } satisfies OrchestrationIntegrationHarness; }); diff --git a/apps/server/integration/TransferBudgetScenario.integration.ts b/apps/server/integration/TransferBudgetScenario.integration.ts index 0807481abd1..77dfbc1dd7f 100644 --- a/apps/server/integration/TransferBudgetScenario.integration.ts +++ b/apps/server/integration/TransferBudgetScenario.integration.ts @@ -47,6 +47,7 @@ const waitForTurnQuiesced = Effect.fn("TransferBudget.waitForTurnQuiesced")(func receipt.checkpointTurnCount === checkpointTurnCount, ); yield* harness.drainProviderRuntime; + yield* harness.drainCheckpointReactor; return receipt; }); diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index d2c2ee89593..4ddb01e09dd 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -102,7 +102,7 @@ const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; import * as ServerConfig from "./config.ts"; import { makeRoutesLayer } from "./server.ts"; -import { resolveAvailableEditorsForConfig } from "./ws.ts"; +import { isThreadDetailEvent, resolveAvailableEditorsForConfig } from "./ws.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as GitManager from "./git/GitManager.ts"; import * as Keybindings from "./keybindings.ts"; @@ -7987,12 +7987,24 @@ it.live( createdAt: TRANSFER_MEASURED_TURN_CREATED_AT, }); yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1); - const finalSequence = yield* harness.engine.latestSequence; + const finalThreadSequence = yield* harness.engine + .readEvents(decodedThread.snapshotSequence, 10_000) + .pipe( + Stream.runFold( + () => decodedThread.snapshotSequence, + (sequence, event) => + event.aggregateId === TRANSFER_THREAD_ID && isThreadDetailEvent(event) + ? Math.max(sequence, event.sequence) + : sequence, + ), + ); + assert.isAbove(finalThreadSequence, decodedThread.snapshotSequence); yield* collectQueueUntil( threadItems, - (item) => item.kind === "event" && item.event.sequence >= finalSequence, - `${provider} thread stream to reach sequence ${finalSequence}`, + (item) => + item.kind === "event" && item.event.sequence === finalThreadSequence, + `${provider} thread stream to reach sequence ${finalThreadSequence}`, ); const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals()); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 6bafb9ec3ba..a6b155c296f 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -268,7 +268,7 @@ function projectSetupScriptCompatibilityDetail( } } -function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< +export function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< OrchestrationEvent, { type: From c9729d2aab3b375be201e7c0e822be5d23816fd7 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 22:20:18 -0700 Subject: [PATCH 07/10] fix(ci): avoid ambiguous transfer report targets --- .github/scripts/thread-transfer-report.cjs | 16 +++++- .../scripts/thread-transfer-report.test.cjs | 55 +++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/.github/scripts/thread-transfer-report.cjs b/.github/scripts/thread-transfer-report.cjs index da14698ac97..2ff75f69be5 100644 --- a/.github/scripts/thread-transfer-report.cjs +++ b/.github/scripts/thread-transfer-report.cjs @@ -243,7 +243,21 @@ async function resolve({ github, context, core }) { github.rest.repos.listPullRequestsAssociatedWithCommit, { owner, repo, commit_sha: source.head_sha, per_page: 100 }, ); - pullNumber = associated.find((pull) => pull.state === "open")?.number; + const matchingPulls = associated.filter( + (pull) => + pull.state === "open" && + pull.head.sha === source.head_sha && + pull.head.ref === source.head_branch && + pull.head.repo?.full_name === source.head_repository?.full_name, + ); + if (matchingPulls.length !== 1) { + core.info( + `Expected one open pull request for ${source.head_repository?.full_name ?? "unknown repository"}:${source.head_branch ?? "unknown branch"} at ${source.head_sha}; found ${matchingPulls.length}.`, + ); + core.setOutput("publish", "false"); + return; + } + pullNumber = matchingPulls[0].number; } if (!pullNumber) { core.info("No open pull request is associated with the completed CI run."); diff --git a/.github/scripts/thread-transfer-report.test.cjs b/.github/scripts/thread-transfer-report.test.cjs index a6a3c40e420..5ae3e69dcc8 100644 --- a/.github/scripts/thread-transfer-report.test.cjs +++ b/.github/scripts/thread-transfer-report.test.cjs @@ -145,6 +145,61 @@ test("resolves the current PR artifact and exact main baseline", async () => { assert.equal(outputs.baseline_matches_base, "true"); }); +test("does not guess when a fallback commit belongs to multiple PRs", async () => { + const outputs = {}; + const listPullRequestsAssociatedWithCommit = () => {}; + let fetchedPull = false; + await resolve({ + github: { + paginate: async (method) => { + assert.equal(method, listPullRequestsAssociatedWithCommit); + return [5350, 5351].map((number) => ({ + number, + state: "open", + head: { + sha: "head-sha", + ref: "feature-branch", + repo: { full_name: "pingdotgg/t3code" }, + }, + })); + }, + rest: { + actions: {}, + pulls: { + get: async () => { + fetchedPull = true; + }, + }, + repos: { listPullRequestsAssociatedWithCommit }, + }, + }, + context: { + repo: { owner: "pingdotgg", repo: "t3code" }, + payload: { + workflow_run: { + id: 2, + event: "pull_request", + workflow_id: 3, + head_sha: "head-sha", + head_branch: "feature-branch", + head_repository: { full_name: "pingdotgg/t3code" }, + conclusion: "success", + pull_requests: [], + }, + }, + }, + core: { + info: () => {}, + setOutput: (key, value) => { + outputs[key] = value; + }, + }, + }); + + assert.equal(outputs.publish, "false"); + assert.equal(fetchedPull, false); +}); + test("does not publish a stale result after the PR head advances", async () => { let listedComments = false; const info = []; From 6f0e9d8ecaa0042d92e5be92584d807909157d21 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 22:23:38 -0700 Subject: [PATCH 08/10] fix(ci): preserve successful transfer reports --- .github/scripts/thread-transfer-report.cjs | 26 +++++++++-- .../scripts/thread-transfer-report.test.cjs | 43 +++++++++++++++++++ 2 files changed, 66 insertions(+), 3 deletions(-) diff --git a/.github/scripts/thread-transfer-report.cjs b/.github/scripts/thread-transfer-report.cjs index 2ff75f69be5..92c8f0c8ac8 100644 --- a/.github/scripts/thread-transfer-report.cjs +++ b/.github/scripts/thread-transfer-report.cjs @@ -29,6 +29,10 @@ const SCENARIO_KEYS = [ "measuredMcpResultBytes", ]; +function resultShaMarker(sha) { + return ``; +} + function assertObject(value, label) { if (value === null || typeof value !== "object" || Array.isArray(value)) { throw new Error(`${label} must be an object`); @@ -187,6 +191,7 @@ function renderComment(input) { return [ COMMENT_MARKER, + resultShaMarker(input.currentRun.sha), "## Thread transfer impact", "", failed @@ -312,7 +317,7 @@ async function resolve({ github, context, core }) { ); } -async function upsertComment(github, context, pullNumber, body) { +async function upsertComment(github, context, pullNumber, body, options = {}) { const { owner, repo } = context.repo; const comments = await github.paginate(github.rest.issues.listComments, { owner, @@ -324,6 +329,12 @@ async function upsertComment(github, context, pullNumber, body) { (comment) => comment.user?.login === "github-actions[bot]" && comment.body?.includes(COMMENT_MARKER), ); + if ( + options.preserveResultSha && + existing?.body?.includes(resultShaMarker(options.preserveResultSha)) + ) { + return; + } if (existing) { await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body }); } else { @@ -331,7 +342,15 @@ async function upsertComment(github, context, pullNumber, body) { } } -async function upsertCommentForCurrentHead(github, context, core, pullNumber, expectedSha, body) { +async function upsertCommentForCurrentHead( + github, + context, + core, + pullNumber, + expectedSha, + body, + options, +) { const { owner, repo } = context.repo; const { data: pull } = await github.rest.pulls.get({ owner, @@ -343,7 +362,7 @@ async function upsertCommentForCurrentHead(github, context, core, pullNumber, ex return false; } - await upsertComment(github, context, pullNumber, body); + await upsertComment(github, context, pullNumber, body, options); return true; } @@ -374,6 +393,7 @@ async function publish({ github, context, core }) { "", "_This comment will update automatically after the next completed run._", ].join("\n"), + { preserveResultSha: currentRun.sha }, ); return; } diff --git a/.github/scripts/thread-transfer-report.test.cjs b/.github/scripts/thread-transfer-report.test.cjs index 5ae3e69dcc8..1a9049fbb98 100644 --- a/.github/scripts/thread-transfer-report.test.cjs +++ b/.github/scripts/thread-transfer-report.test.cjs @@ -80,6 +80,10 @@ test("renders baseline, impact, ceiling, and ceiling changes", () => { assert.match(comment, /This PR changes transfer ceilings/); assert.match(comment, /312\.5 KiB → 322\.3 KiB/); assert.match(comment, //); + assert.match( + comment, + //, + ); }); test("resolves the current PR artifact and exact main baseline", async () => { @@ -235,3 +239,42 @@ test("does not publish a stale result after the PR head advances", async () => { assert.equal(listedComments, false); assert.deepEqual(info, ["Skipping stale CI result old-head-sha; PR head is new-head-sha."]); }); + +test("preserves a successful result when a same-SHA rerun has no artifact", async () => { + let updatedComment = false; + const sha = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + const published = await upsertCommentForCurrentHead( + { + paginate: async () => [ + { + id: 1, + user: { login: "github-actions[bot]" }, + body: `\n`, + }, + ], + rest: { + issues: { + listComments: () => {}, + createComment: () => { + updatedComment = true; + }, + updateComment: () => { + updatedComment = true; + }, + }, + pulls: { + get: async () => ({ data: { head: { sha } } }), + }, + }, + }, + { repo: { owner: "pingdotgg", repo: "t3code" } }, + { info: () => {} }, + 5350, + sha, + "missing artifact warning", + { preserveResultSha: sha }, + ); + + assert.equal(published, true); + assert.equal(updatedComment, false); +}); From 9347584fd3dfdc410ff591f532b87351c480e390 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 4 Aug 2026 22:28:37 -0700 Subject: [PATCH 09/10] fix(ci): resolve fork transfer reports --- .github/scripts/thread-transfer-report.cjs | 3 +-- .../scripts/thread-transfer-report.test.cjs | 18 +++++++++++++++--- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/.github/scripts/thread-transfer-report.cjs b/.github/scripts/thread-transfer-report.cjs index 92c8f0c8ac8..94a02b7806d 100644 --- a/.github/scripts/thread-transfer-report.cjs +++ b/.github/scripts/thread-transfer-report.cjs @@ -252,8 +252,7 @@ async function resolve({ github, context, core }) { (pull) => pull.state === "open" && pull.head.sha === source.head_sha && - pull.head.ref === source.head_branch && - pull.head.repo?.full_name === source.head_repository?.full_name, + pull.head.ref === source.head_branch, ); if (matchingPulls.length !== 1) { core.info( diff --git a/.github/scripts/thread-transfer-report.test.cjs b/.github/scripts/thread-transfer-report.test.cjs index 1a9049fbb98..4935864e46f 100644 --- a/.github/scripts/thread-transfer-report.test.cjs +++ b/.github/scripts/thread-transfer-report.test.cjs @@ -86,12 +86,22 @@ test("renders baseline, impact, ceiling, and ceiling changes", () => { ); }); -test("resolves the current PR artifact and exact main baseline", async () => { +test("resolves a fallback PR with a redacted head repo and exact main baseline", async () => { const outputs = {}; const listWorkflowRunArtifacts = () => {}; const listWorkflowRuns = () => {}; + const listPullRequestsAssociatedWithCommit = () => {}; const github = { paginate: async (method, input) => { + if (method === listPullRequestsAssociatedWithCommit) { + return [ + { + number: 5350, + state: "open", + head: { sha: "head-sha", ref: "feature-branch", repo: null }, + }, + ]; + } if (method === listWorkflowRunArtifacts) { return [ { @@ -116,7 +126,7 @@ test("resolves the current PR artifact and exact main baseline", async () => { }, }), }, - repos: { listPullRequestsAssociatedWithCommit: () => {} }, + repos: { listPullRequestsAssociatedWithCommit }, }, }; await resolve({ @@ -129,8 +139,10 @@ test("resolves the current PR artifact and exact main baseline", async () => { event: "pull_request", workflow_id: 3, head_sha: "head-sha", + head_branch: "feature-branch", + head_repository: { full_name: "pingdotgg/t3code" }, conclusion: "success", - pull_requests: [{ number: 5350 }], + pull_requests: [], }, }, }, From 344bc61d56c93e1a9cb94f6e7d5895953c93f981 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 22:58:19 -0700 Subject: [PATCH 10/10] test(server): recalibrate thread transfer budgets --- .../TransferBudgetReport.integration.ts | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index ba5166f48c1..f773b5b8b84 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -26,15 +26,16 @@ interface ProviderTransferBudget { readonly measuredTurnWebSocketMessages: number; } -// These caps leave roughly 30% headroom above the deterministic fixture. The -// CI report preserves the exact values so intentional protocol growth can be -// reviewed and the caps raised explicitly. +// These caps leave roughly 30% headroom above the client projection of the +// deterministic 9 MB retained-result fixture. Full MCP results stay in +// persistence, so accidentally shipping them again exceeds these caps by +// orders of magnitude. The CI report preserves exact values for review. const TRANSFER_BUDGET = { - totalWireBytes: 2_900_000, - threadSnapshotWireBytes: 2_600_000, - measuredTurnWebSocketWireBytes: 320_000, - measuredTurnWebSocketDecodedBytes: 1_550_000, - measuredTurnWebSocketMessages: 20, + totalWireBytes: 15_500, + threadSnapshotWireBytes: 7_500, + measuredTurnWebSocketWireBytes: 8_000, + measuredTurnWebSocketDecodedBytes: 68_000, + measuredTurnWebSocketMessages: 21, } satisfies ProviderTransferBudget; export const TRANSFER_BUDGETS: Readonly> = {