authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-01 19:16:09-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 19:16:09-07:00
logc37fd136edf546fd58daeac146548436c1fa38c2
treef7e56ca1dcdede8c798700a8486599bc662ca61a
parentb2f25d390a8546e42eb6186115b673a53b8c38dc
signature Signed by PGP key B5690EEEBB952194

test(server): measure shell, second client, and reconnect transfer (#9157)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

6 files changed, 436 insertions(+), 65 deletions(-)

apps/server/integration/NetworkTransferMeasurement.integration.ts+59-1
...@@ -5,8 +5,10 @@ import * as NodeZlib from "node:zlib";...@@ -5,8 +5,10 @@ import * as NodeZlib from "node:zlib";
5import * as NodeSocket from "@effect/platform-node/NodeSocket";5import * as NodeSocket from "@effect/platform-node/NodeSocket";
6import { WsRpcGroup } from "@t3tools/contracts";6import { WsRpcGroup } from "@t3tools/contracts";
7import * as Effect from "effect/Effect";7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";
8import * as Layer from "effect/Layer";9import * as Layer from "effect/Layer";
9import * as Schema from "effect/Schema";10import * as Schema from "effect/Schema";
11import * as Scope from "effect/Scope";
10import { RpcClient, RpcSerialization } from "effect/unstable/rpc";12import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
11import * as Socket from "effect/unstable/socket/Socket";13import * as Socket from "effect/unstable/socket/Socket";
1214
...@@ -101,6 +103,8 @@ export interface WebSocketTransferRecorder {...@@ -101,6 +103,8 @@ export interface WebSocketTransferRecorder {
101 ) => globalThis.WebSocket;103 ) => globalThis.WebSocket;
102 readonly totals: () => WebSocketTransferTotals;104 readonly totals: () => WebSocketTransferTotals;
103 readonly negotiatedExtensions: () => string;105 readonly negotiatedExtensions: () => string;
106 /** Resolves once the upgrade completes, so totals taken after it exclude the upgrade response. */
107 readonly awaitOpen: Effect.Effect<void>;
104}108}
105109
106interface NodeWebSocketWithTransport extends NodeSocket.NodeWS.WebSocket {110interface NodeWebSocketWithTransport extends NodeSocket.NodeWS.WebSocket {
...@@ -118,8 +122,15 @@ function rawDataBytes(data: NodeSocket.NodeWS.RawData): number {...@@ -118,8 +122,15 @@ function rawDataBytes(data: NodeSocket.NodeWS.RawData): number {
118122
119export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {123export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {
120 let socket: NodeWebSocketWithTransport | null = null;124 let socket: NodeWebSocketWithTransport | null = null;
125 // Held separately from the WebSocket so wire totals survive a close, which
126 // is when a reconnect measurement reads them.
127 let transport: NodeWebSocketWithTransport["_socket"] | null = null;
121 let decodedBytes = 0;128 let decodedBytes = 0;
122 let messages = 0;129 let messages = 0;
130 let resolveOpen: () => void = () => {};
131 const opened = new Promise<void>((resolve) => {
132 resolveOpen = resolve;
133 });
123134
124 return {135 return {
125 connect: (url, protocols, cookie) => {136 connect: (url, protocols, cookie) => {
...@@ -128,6 +139,10 @@ export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {...@@ -128,6 +139,10 @@ export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {
128 perMessageDeflate: true,139 perMessageDeflate: true,
129 }) as NodeWebSocketWithTransport;140 }) as NodeWebSocketWithTransport;
130 socket = nextSocket;141 socket = nextSocket;
142 nextSocket.once("open", () => {
143 transport = nextSocket._socket ?? null;
144 resolveOpen();
145 });
131 nextSocket.on("message", (data) => {146 nextSocket.on("message", (data) => {
132 const bytes = rawDataBytes(data);147 const bytes = rawDataBytes(data);
133 decodedBytes += bytes;148 decodedBytes += bytes;
...@@ -136,11 +151,17 @@ export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {...@@ -136,11 +151,17 @@ export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {
136 return nextSocket as unknown as globalThis.WebSocket;151 return nextSocket as unknown as globalThis.WebSocket;
137 },152 },
138 totals: () => ({153 totals: () => ({
139 wireBytes: socket?._socket?.bytesRead ?? 0,154 wireBytes: transport?.bytesRead ?? socket?._socket?.bytesRead ?? 0,
140 decodedBytes,155 decodedBytes,
141 messages,156 messages,
142 }),157 }),
143 negotiatedExtensions: () => socket?.extensions ?? "",158 negotiatedExtensions: () => socket?.extensions ?? "",
159 awaitOpen: Effect.promise(() => opened).pipe(
160 Effect.timeoutOrElse({
161 duration: "10 seconds",
162 orElse: () => Effect.die(new Error("Timed out waiting for the WebSocket to open")),
163 }),
164 ),
144 };165 };
145}166}
146167
...@@ -175,3 +196,40 @@ export function countingWsRpcProtocolLayer(input: {...@@ -175,3 +196,40 @@ export function countingWsRpcProtocolLayer(input: {
175196
176export const makeCountingWsRpcClient = RpcClient.make(WsRpcGroup);197export const makeCountingWsRpcClient = RpcClient.make(WsRpcGroup);
177export type CountingWsRpcClient = Effect.Success<typeof makeCountingWsRpcClient>;198export type CountingWsRpcClient = Effect.Success<typeof makeCountingWsRpcClient>;
199
200export interface MeasuredWsClient {
201 readonly client: CountingWsRpcClient;
202 readonly recorder: WebSocketTransferRecorder;
203 /** Fork subscription consumers here so they stop before the socket closes. */
204 readonly scope: Scope.Scope;
205 /** Closes the socket now. The enclosing scope closes it otherwise. */
206 readonly close: Effect.Effect<void>;
207}
208
209/**
210 * Opens one WebSocket RPC client on a child of the current scope. Several
211 * clients can share one test scope and still disconnect independently, which
212 * a reconnect measurement needs.
213 */
214export const openMeasuredWsClient = Effect.fn("TransferBudget.openMeasuredWsClient")(
215 function* (input: { readonly url: string; readonly cookie: string }) {
216 const recorder = makeWebSocketTransferRecorder();
217 const parent = yield* Effect.scope;
218 const scope = yield* Scope.fork(parent);
219 const protocol = yield* Layer.buildWithScope(
220 countingWsRpcProtocolLayer({ url: input.url, cookie: input.cookie, recorder }),
221 scope,
222 );
223 const client = yield* makeCountingWsRpcClient.pipe(
224 Effect.provide(protocol),
225 Scope.provide(scope),
226 );
227 yield* recorder.awaitOpen;
228 return {
229 client,
230 recorder,
231 scope,
232 close: Scope.close(scope, Exit.void),
233 } satisfies MeasuredWsClient;
234 },
235);
apps/server/integration/OrchestrationEngineHarness.integration.ts+6
...@@ -22,6 +22,7 @@ import * as Schedule from "effect/Schedule";...@@ -22,6 +22,7 @@ import * as Schedule from "effect/Schedule";
22import * as Schema from "effect/Schema";22import * as Schema from "effect/Schema";
23import * as Scope from "effect/Scope";23import * as Scope from "effect/Scope";
24import * as Stream from "effect/Stream";24import * as Stream from "effect/Stream";
25import * as Tracer from "effect/Tracer";
2526
26import * as CheckpointStore from "../src/checkpointing/CheckpointStore.ts";27import * as CheckpointStore from "../src/checkpointing/CheckpointStore.ts";
27import { TextGeneration, type TextGenerationShape } from "../src/textGeneration/TextGeneration.ts";28import { TextGeneration, type TextGenerationShape } from "../src/textGeneration/TextGeneration.ts";
...@@ -231,6 +232,8 @@ export interface OrchestrationIntegrationHarness {...@@ -231,6 +232,8 @@ export interface OrchestrationIntegrationHarness {
231interface MakeOrchestrationIntegrationHarnessOptions {232interface MakeOrchestrationIntegrationHarnessOptions {
232 readonly provider?: ProviderDriverKind;233 readonly provider?: ProviderDriverKind;
233 readonly realCodex?: boolean;234 readonly realCodex?: boolean;
235 /** Tracer for every fiber the harness runtime runs, including reactors. */
236 readonly tracer?: Tracer.Tracer;
234}237}
235238
236export const makeOrchestrationIntegrationHarness = (239export const makeOrchestrationIntegrationHarness = (
...@@ -399,6 +402,9 @@ export const makeOrchestrationIntegrationHarness = (...@@ -399,6 +402,9 @@ export const makeOrchestrationIntegrationHarness = (
399 Layer.provideMerge(ServerSettingsService.layerTest()),402 Layer.provideMerge(ServerSettingsService.layerTest()),
400 Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),403 Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),
401 Layer.provideMerge(NodeServices.layer),404 Layer.provideMerge(NodeServices.layer),
405 Layer.provideMerge(
406 options?.tracer ? Layer.succeed(Tracer.Tracer, options.tracer) : Layer.empty,
407 ),
402 );408 );
403409
404 const runtime = ManagedRuntime.make(layer);410 const runtime = ManagedRuntime.make(layer);
apps/server/integration/SqlStatementCounter.integration.ts created+24
...@@ -0,0 +1,24 @@
1import * as Tracer from "effect/Tracer";
2
3export interface SqlStatementCounter {
4 readonly tracer: Tracer.Tracer;
5 /** Statements executed so far. Read before and after a phase and subtract. */
6 readonly count: () => number;
7}
8
9/**
10 * Counts `sql.execute` spans, which the Effect SQL client opens once per
11 * statement. Spans behave exactly as with the native tracer. Install the same
12 * counter on every runtime under test, otherwise statements run by background
13 * reactors and statements run by request handlers land in different counters.
14 */
15export function makeSqlStatementCounter(): SqlStatementCounter {
16 let statements = 0;
17 const tracer = Tracer.make({
18 span: (options) => {
19 if (options.name === "sql.execute") statements += 1;
20 return new Tracer.NativeSpan(options);
21 },
22 });
23 return { tracer, count: () => statements };
24}
apps/server/integration/TransferBudgetReport.integration.ts+82
...@@ -12,10 +12,28 @@ import {...@@ -12,10 +12,28 @@ import {
12 TRANSFER_MEASURED_TOOLS,12 TRANSFER_MEASURED_TOOLS,
13} from "./fixtures/transferBudget.ts";13} from "./fixtures/transferBudget.ts";
1414
15/** Catch-up delivered to a resubscribing client, and which path the server chose. */
16export interface WebSocketCatchUpMeasurement extends WebSocketTransferTotals {
17 readonly mode: "replay" | "snapshot";
18}
19
15export interface TransferBudgetRun {20export interface TransferBudgetRun {
16 readonly provider: ProviderDriverKind;21 readonly provider: ProviderDriverKind;
17 readonly threadSnapshot: HttpTransferMeasurement;22 readonly threadSnapshot: HttpTransferMeasurement;
23 /** One socket holding only the thread subscription. This is the capped measurement. */
18 readonly measuredTurnWebSocket: WebSocketTransferTotals;24 readonly measuredTurnWebSocket: WebSocketTransferTotals;
25 readonly shellSnapshot: HttpTransferMeasurement;
26 /** One socket holding only the shell (sidebar) subscription during the same turn. */
27 readonly measuredTurnShellWebSocket: WebSocketTransferTotals;
28 /** A second client: one socket holding both the thread and shell subscriptions. */
29 readonly measuredTurnSecondClientWebSocket: WebSocketTransferTotals;
30 /** The second client resubscribes after the turn from the cursor it held before it. */
31 readonly reconnectThread: WebSocketCatchUpMeasurement;
32 readonly reconnectShell: WebSocketCatchUpMeasurement;
33 /** `sql.execute` spans opened server-wide during the measured turn. */
34 readonly measuredTurnSqlStatements: number;
35 /** `sql.execute` spans opened while serving both reconnect catch-ups. */
36 readonly reconnectSqlStatements: number;
19}37}
2038
21interface ProviderTransferBudget {39interface ProviderTransferBudget {
...@@ -47,6 +65,15 @@ function totalWireBytes(run: TransferBudgetRun): number {...@@ -47,6 +65,15 @@ function totalWireBytes(run: TransferBudgetRun): number {
47 return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes;65 return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes;
48}66}
4967
68/** Bytes the server wrote to every measured socket during the turn. */
69function serverEgressWireBytes(run: TransferBudgetRun): number {
70 return (
71 run.measuredTurnWebSocket.wireBytes +
72 run.measuredTurnShellWebSocket.wireBytes +
73 run.measuredTurnSecondClientWebSocket.wireBytes
74 );
75}
76
50function observedTransfer(run: TransferBudgetRun) {77function observedTransfer(run: TransferBudgetRun) {
51 return {78 return {
52 totalWireBytes: totalWireBytes(run),79 totalWireBytes: totalWireBytes(run),
...@@ -105,6 +132,33 @@ function row(...@@ -105,6 +132,33 @@ function row(
105 return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | ${format(maximum)} | ${status} |`;132 return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | ${format(maximum)} | ${status} |`;
106}133}
107134
135// Shell, second-client, reconnect, and SQL rows are reported without a cap.
136// Shell delivery coalesces on a 50 ms window, so message counts and bytes move
137// with scheduler timing between runs, and the reconnect and SQL figures follow
138// the same batches. The rows exist so CI shows the numbers next to the capped
139// thread measurement.
140function infoRow(
141 provider: ProviderDriverKind,
142 phase: string,
143 metric: string,
144 observed: number,
145 format: (value: number) => string = formatBytes,
146): string {
147 return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | none | INFO |`;
148}
149
150function webSocketRows(
151 provider: ProviderDriverKind,
152 phase: string,
153 totals: WebSocketTransferTotals,
154): string[] {
155 return [
156 infoRow(provider, phase, "WebSocket wire", totals.wireBytes),
157 infoRow(provider, phase, "WebSocket decoded", totals.decodedBytes),
158 infoRow(provider, phase, "WebSocket messages", totals.messages, String),
159 ];
160}
161
108export function transferBudgetViolations(runs: ReadonlyArray<TransferBudgetRun>): string[] {162export function transferBudgetViolations(runs: ReadonlyArray<TransferBudgetRun>): string[] {
109 const violations: string[] = [];163 const violations: string[] = [];
110 for (const run of runs) {164 for (const run of runs) {
...@@ -146,6 +200,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun...@@ -146,6 +200,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun
146 "# T3 Code thread transfer budget",200 "# T3 Code thread transfer budget",
147 "",201 "",
148 "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.",202 "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.",
203 "The measured turn is observed on three sockets at once: one with only the thread subscription (the capped rows), one with only the shell subscription, and a second client holding both. Server egress is the sum of the three. After the turn the second client disconnects and resubscribes from the cursor it held before the turn, which is the cursor a backgrounded phone would hold. SQL statements are `sql.execute` spans counted across the orchestration runtime and the WebSocket handlers.",
149 `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.`,204 `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.`,
150 "",205 "",
151 "| Provider | Total thread wire | Budget | Result |",206 "| Provider | Total thread wire | Budget | Result |",
...@@ -198,6 +253,32 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun...@@ -198,6 +253,32 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun
198 budget.measuredTurnWebSocketMessages,253 budget.measuredTurnWebSocketMessages,
199 String,254 String,
200 ),255 ),
256 infoRow(run.provider, "shell snapshot", "HTTP wire", run.shellSnapshot.wireBytes),
257 ...webSocketRows(run.provider, "measured turn, shell", run.measuredTurnShellWebSocket),
258 ...webSocketRows(
259 run.provider,
260 "measured turn, second client",
261 run.measuredTurnSecondClientWebSocket,
262 ),
263 infoRow(run.provider, "measured turn", "server egress wire", serverEgressWireBytes(run)),
264 infoRow(
265 run.provider,
266 "measured turn",
267 "SQL statements",
268 run.measuredTurnSqlStatements,
269 String,
270 ),
271 ...webSocketRows(
272 run.provider,
273 `reconnect, thread (${run.reconnectThread.mode})`,
274 run.reconnectThread,
275 ),
276 ...webSocketRows(
277 run.provider,
278 `reconnect, shell (${run.reconnectShell.mode})`,
279 run.reconnectShell,
280 ),
281 infoRow(run.provider, "reconnect", "SQL statements", run.reconnectSqlStatements, String),
201 );282 );
202 }283 }
203284
...@@ -205,6 +286,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun...@@ -205,6 +286,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun
205 for (const run of runs) {286 for (const run of runs) {
206 lines.push(287 lines.push(
207 `- ${run.provider}: thread snapshot ${formatBytes(run.threadSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.threadSnapshot.encodedBodyBytes)} gzip.`,288 `- ${run.provider}: thread snapshot ${formatBytes(run.threadSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.threadSnapshot.encodedBodyBytes)} gzip.`,
289 `- ${run.provider}: shell snapshot ${formatBytes(run.shellSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.shellSnapshot.encodedBodyBytes)} gzip.`,
208 );290 );
209 }291 }
210292
apps/server/integration/TransferBudgetScenario.integration.ts+84
...@@ -5,13 +5,20 @@ import {...@@ -5,13 +5,20 @@ import {
5 DEFAULT_MODEL_BY_PROVIDER,5 DEFAULT_MODEL_BY_PROVIDER,
6 DEFAULT_PROVIDER_INTERACTION_MODE,6 DEFAULT_PROVIDER_INTERACTION_MODE,
7 MessageId,7 MessageId,
8 ORCHESTRATION_WS_METHODS,
9 type OrchestrationShellStreamItem,
10 type OrchestrationThreadStreamItem,
8 ProjectId,11 ProjectId,
9 ProviderDriverKind,12 ProviderDriverKind,
10 ThreadId,13 ThreadId,
11} from "@t3tools/contracts";14} from "@t3tools/contracts";
12import * as Effect from "effect/Effect";15import * as Effect from "effect/Effect";
16import * as Queue from "effect/Queue";
17import * as Scope from "effect/Scope";
18import * as Stream from "effect/Stream";
1319
14import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts";20import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts";
21import type { MeasuredWsClient } from "./NetworkTransferMeasurement.integration.ts";
15import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts";22import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts";
16import {23import {
17 expectedRecordedAssistantText,24 expectedRecordedAssistantText,
...@@ -125,4 +132,81 @@ export function expectedMeasuredAssistantText(provider: ProviderDriverKind): str...@@ -125,4 +132,81 @@ export function expectedMeasuredAssistantText(provider: ProviderDriverKind): str
125 return expectedRecordedAssistantText(provider, TRANSFER_MEASURED_TURN_INDEX);132 return expectedRecordedAssistantText(provider, TRANSFER_MEASURED_TURN_INDEX);
126}133}
127134
135/** Takes from the queue until one value matches, and returns everything taken. */
136export const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function* <A>(
137 queue: Queue.Queue<A>,
138 predicate: (value: A) => boolean,
139 waitDescription: string,
140) {
141 return yield* Effect.gen(function* () {
142 const values: A[] = [];
143 while (true) {
144 const value = yield* Queue.take(queue);
145 values.push(value);
146 if (predicate(value)) return values;
147 }
148 }).pipe(
149 Effect.timeoutOrElse({
150 duration: "10 seconds",
151 orElse: () => Effect.die(new Error(`Timed out waiting for ${waitDescription}`)),
152 }),
153 );
154});
155
156/**
157 * Resumes the thread subscription from a cursor on a measured client. The
158 * consumer runs in the client's scope, so closing the client stops it. Callers
159 * read items from the returned queue.
160 */
161export const subscribeThreadItems = Effect.fn("TransferBudget.subscribeThreadItems")(function* (
162 measured: MeasuredWsClient,
163 afterSequence: number,
164) {
165 const items = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
166 yield* measured.client[ORCHESTRATION_WS_METHODS.subscribeThread]({
167 threadId: TRANSFER_THREAD_ID,
168 afterSequence,
169 requestCompletionMarker: true,
170 }).pipe(
171 Stream.runForEach((item) => Queue.offer(items, item).pipe(Effect.asVoid)),
172 Scope.provide(measured.scope),
173 Effect.forkIn(measured.scope),
174 );
175 return items;
176});
177
178/** Shell counterpart of subscribeThreadItems. */
179export const subscribeShellItems = Effect.fn("TransferBudget.subscribeShellItems")(function* (
180 measured: MeasuredWsClient,
181 afterSequence: number,
182) {
183 const items = yield* Queue.unbounded<OrchestrationShellStreamItem>();
184 yield* measured.client[ORCHESTRATION_WS_METHODS.subscribeShell]({
185 afterSequence,
186 requestCompletionMarker: true,
187 }).pipe(
188 Stream.runForEach((item) => Queue.offer(items, item).pipe(Effect.asVoid)),
189 Scope.provide(measured.scope),
190 Effect.forkIn(measured.scope),
191 );
192 return items;
193});
194
195/** Waits for the initial catch-up and reports whether it was a replay or a snapshot reset. */
196export const awaitSubscriptionSynchronized = Effect.fn(
197 "TransferBudget.awaitSubscriptionSynchronized",
198)(function* <Item extends OrchestrationThreadStreamItem | OrchestrationShellStreamItem>(
199 items: Queue.Queue<Item>,
200 waitDescription: string,
201) {
202 const initial = yield* collectQueueUntil(
203 items,
204 (item) => item.kind === "synchronized",
205 waitDescription,
206 );
207 return initial.some((item) => item.kind === "snapshot")
208 ? ("snapshot" as const)
209 : ("replay" as const);
210});
211
128export { TRANSFER_HISTORY_TURN_COUNT, waitForTurnQuiesced };212export { TRANSFER_HISTORY_TURN_COUNT, waitForTurnQuiesced };
apps/server/src/server.test.ts+181-64
...@@ -17,6 +17,8 @@ import {...@@ -17,6 +17,8 @@ import {
17 KeybindingRule,17 KeybindingRule,
18 MessageId,18 MessageId,
19 ExternalLauncherCommandNotFoundError,19 ExternalLauncherCommandNotFoundError,
20 OrchestrationShellSnapshot,
21 type OrchestrationShellStreamItem,
20 OrchestrationThreadDetailSnapshot,22 OrchestrationThreadDetailSnapshot,
21 type OrchestrationThreadStreamItem,23 type OrchestrationThreadStreamItem,
22 type OrchestrationThreadActivity,24 type OrchestrationThreadActivity,
...@@ -58,11 +60,11 @@ import * as ManagedRuntime from "effect/ManagedRuntime";...@@ -58,11 +60,11 @@ import * as ManagedRuntime from "effect/ManagedRuntime";
58import * as Option from "effect/Option";60import * as Option from "effect/Option";
59import * as Path from "effect/Path";61import * as Path from "effect/Path";
60import * as PubSub from "effect/PubSub";62import * as PubSub from "effect/PubSub";
61import * as Queue from "effect/Queue";
62import * as Ref from "effect/Ref";63import * as Ref from "effect/Ref";
63import * as Schema from "effect/Schema";64import * as Schema from "effect/Schema";
64import * as Stream from "effect/Stream";65import * as Stream from "effect/Stream";
65import * as TestClock from "effect/testing/TestClock";66import * as TestClock from "effect/testing/TestClock";
67import * as Tracer from "effect/Tracer";
66import { ChildProcessSpawner } from "effect/unstable/process";68import { ChildProcessSpawner } from "effect/unstable/process";
67import {69import {
68 FetchHttpClient,70 FetchHttpClient,
...@@ -82,26 +84,9 @@ const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z");...@@ -82,26 +84,9 @@ const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z");
82const decodeTransferThreadSnapshot = Schema.decodeUnknownEffect(84const decodeTransferThreadSnapshot = Schema.decodeUnknownEffect(
83 Schema.fromJsonString(OrchestrationThreadDetailSnapshot),85 Schema.fromJsonString(OrchestrationThreadDetailSnapshot),
84);86);
8587const decodeTransferShellSnapshot = Schema.decodeUnknownEffect(
86const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function* <A>(88 Schema.fromJsonString(OrchestrationShellSnapshot),
87 queue: Queue.Queue<A>,89);
88 predicate: (value: A) => boolean,
89 waitDescription: string,
90) {
91 return yield* Effect.gen(function* () {
92 const values: A[] = [];
93 while (true) {
94 const value = yield* Queue.take(queue);
95 values.push(value);
96 if (predicate(value)) return values;
97 }
98 }).pipe(
99 Effect.timeoutOrElse({
100 duration: "10 seconds",
101 orElse: () => Effect.die(new Error(`Timed out waiting for ${waitDescription}`)),
102 }),
103 );
104});
10590
106import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";91import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";
107import * as ServerConfig from "./config.ts";92import * as ServerConfig from "./config.ts";
...@@ -171,16 +156,19 @@ import * as Data from "effect/Data";...@@ -171,16 +156,19 @@ import * as Data from "effect/Data";
171156
172import { makeOrchestrationIntegrationHarness } from "../integration/OrchestrationEngineHarness.integration.ts";157import { makeOrchestrationIntegrationHarness } from "../integration/OrchestrationEngineHarness.integration.ts";
173import {158import {
174 countingWsRpcProtocolLayer,
175 makeCountingWsRpcClient,
176 makeWebSocketTransferRecorder,
177 measureHttpGet,159 measureHttpGet,
160 openMeasuredWsClient,
178 transferDelta,161 transferDelta,
179} from "../integration/NetworkTransferMeasurement.integration.ts";162} from "../integration/NetworkTransferMeasurement.integration.ts";
163import { makeSqlStatementCounter } from "../integration/SqlStatementCounter.integration.ts";
180import {164import {
165 awaitSubscriptionSynchronized,
166 collectQueueUntil,
181 expectedMeasuredAssistantText,167 expectedMeasuredAssistantText,
182 queueMeasuredTransferTurn,168 queueMeasuredTransferTurn,
183 seedTransferBudgetHistory,169 seedTransferBudgetHistory,
170 subscribeShellItems,
171 subscribeThreadItems,
184 TRANSFER_HISTORY_TURN_COUNT,172 TRANSFER_HISTORY_TURN_COUNT,
185 TRANSFER_MEASURED_TURN_CREATED_AT,173 TRANSFER_MEASURED_TURN_CREATED_AT,
186 TRANSFER_MEASURED_TURN_INDEX,174 TRANSFER_MEASURED_TURN_INDEX,
...@@ -9313,9 +9301,12 @@ it.live(...@@ -9313,9 +9301,12 @@ it.live(
93139301
9314 const runs = yield* Effect.forEach(9302 const runs = yield* Effect.forEach(
9315 providers,9303 providers,
9316 (provider) =>9304 (provider) => {
9317 Effect.acquireUseRelease(9305 // One counter for the orchestration runtime and the HTTP/WS handlers,
9318 makeOrchestrationIntegrationHarness({ provider }),9306 // so reactor writes and subscription reads land in the same total.
9307 const sqlCounter = makeSqlStatementCounter();
9308 return Effect.acquireUseRelease(
9309 makeOrchestrationIntegrationHarness({ provider, tracer: sqlCounter.tracer }),
9319 (harness) =>9310 (harness) =>
9320 Effect.gen(function* () {9311 Effect.gen(function* () {
9321 yield* seedTransferBudgetHistory(harness, provider);9312 yield* seedTransferBudgetHistory(harness, provider);
...@@ -9328,19 +9319,10 @@ it.live(...@@ -9328,19 +9319,10 @@ it.live(
93289319
9329 const baseUrl = yield* getHttpServerUrl();9320 const baseUrl = yield* getHttpServerUrl();
9330 const cookie = yield* getAuthenticatedSessionCookieHeader();9321 const cookie = yield* getAuthenticatedSessionCookieHeader();
9331
9332 const recorder = makeWebSocketTransferRecorder();
9333 const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws";9322 const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws";
9334 const protocolLayer = countingWsRpcProtocolLayer({
9335 url: wsUrl,
9336 cookie,
9337 recorder,
9338 });
93399323
9340 return yield* Effect.scoped(9324 return yield* Effect.scoped(
9341 Effect.gen(function* () {9325 Effect.gen(function* () {
9342 const client = yield* makeCountingWsRpcClient;
9343
9344 const threadSnapshot = yield* measureHttpGet({9326 const threadSnapshot = yield* measureHttpGet({
9345 url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`,9327 url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`,
9346 headers: { cookie },9328 headers: { cookie },
...@@ -9354,29 +9336,78 @@ it.live(...@@ -9354,29 +9336,78 @@ it.live(
9354 decodedThread.thread.messages.length,9336 decodedThread.thread.messages.length,
9355 TRANSFER_HISTORY_TURN_COUNT * 2,9337 TRANSFER_HISTORY_TURN_COUNT * 2,
9356 );9338 );
9339 const shellSnapshot = yield* measureHttpGet({
9340 url: `${baseUrl}/api/orchestration/shell`,
9341 headers: { cookie },
9342 });
9343 assert.equal(shellSnapshot.status, 200);
9344 const decodedShell = yield* decodeTransferShellSnapshot(
9345 Buffer.from(shellSnapshot.decodedBody).toString("utf8"),
9346 );
9347 assert.equal(decodedShell.threads.length, 1);
9348
9349 // Three sockets, the way real installs look: the capped
9350 // thread-only client, a shell-only socket that isolates the
9351 // sidebar cost, and a second device holding both.
9352 const threadClient = yield* openMeasuredWsClient({ url: wsUrl, cookie });
9353 const shellClient = yield* openMeasuredWsClient({ url: wsUrl, cookie });
9354 const secondClient = yield* openMeasuredWsClient({ url: wsUrl, cookie });
9355 assert.include(
9356 threadClient.recorder.negotiatedExtensions(),
9357 "permessage-deflate",
9358 );
93579359
9358 const threadItems = yield* Queue.unbounded<OrchestrationThreadStreamItem>();9360 const threadItems = yield* subscribeThreadItems(
9359 yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({9361 threadClient,
9360 threadId: TRANSFER_THREAD_ID,9362 decodedThread.snapshotSequence,
9361 afterSequence: decodedThread.snapshotSequence,9363 );
9362 requestCompletionMarker: true,9364 const shellItems = yield* subscribeShellItems(
9363 }).pipe(9365 shellClient,
9364 Stream.runForEach((item) =>9366 decodedShell.snapshotSequence,
9365 Queue.offer(threadItems, item).pipe(Effect.asVoid),9367 );
9368 const secondThreadItems = yield* subscribeThreadItems(
9369 secondClient,
9370 decodedThread.snapshotSequence,
9371 );
9372 const secondShellItems = yield* subscribeShellItems(
9373 secondClient,
9374 decodedShell.snapshotSequence,
9375 );
9376 assert.equal(
9377 yield* awaitSubscriptionSynchronized(
9378 threadItems,
9379 `${provider} thread subscription to synchronize`,
9366 ),9380 ),
9367 Effect.forkScoped,9381 "replay",
9368 );9382 );
9369 const initialThreadItems = yield* collectQueueUntil(9383 assert.equal(
9370 threadItems,9384 yield* awaitSubscriptionSynchronized(
9371 (item) => item.kind === "synchronized",9385 shellItems,
9372 `${provider} thread subscription to synchronize`,9386 `${provider} shell subscription to synchronize`,
9387 ),
9388 "replay",
9389 );
9390 assert.equal(
9391 yield* awaitSubscriptionSynchronized(
9392 secondThreadItems,
9393 `${provider} second client thread subscription to synchronize`,
9394 ),
9395 "replay",
9396 );
9397 assert.equal(
9398 yield* awaitSubscriptionSynchronized(
9399 secondShellItems,
9400 `${provider} second client shell subscription to synchronize`,
9401 ),
9402 "replay",
9373 );9403 );
9374 assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot"));
9375 assert.include(recorder.negotiatedExtensions(), "permessage-deflate");
93769404
9377 yield* queueMeasuredTransferTurn(harness, provider);9405 yield* queueMeasuredTransferTurn(harness, provider);
9378 const turnStartTotals = recorder.totals();9406 const turnStartTotals = threadClient.recorder.totals();
9379 yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({9407 const shellTurnStartTotals = shellClient.recorder.totals();
9408 const secondTurnStartTotals = secondClient.recorder.totals();
9409 const turnStartSqlStatements = sqlCounter.count();
9410 yield* threadClient.client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
9380 type: "thread.turn.start",9411 type: "thread.turn.start",
9381 commandId: CommandId.make(`transfer:${provider}:measured-turn`),9412 commandId: CommandId.make(`transfer:${provider}:measured-turn`),
9382 threadId: TRANSFER_THREAD_ID,9413 threadId: TRANSFER_THREAD_ID,
...@@ -9392,26 +9423,95 @@ it.live(...@@ -9392,26 +9423,95 @@ it.live(
9392 createdAt: TRANSFER_MEASURED_TURN_CREATED_AT,9423 createdAt: TRANSFER_MEASURED_TURN_CREATED_AT,
9393 });9424 });
9394 yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1);9425 yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1);
9395 const finalThreadSequence = yield* harness.engine9426 const finalSequences = yield* harness.engine
9396 .readEvents(decodedThread.snapshotSequence, 10_000)9427 .readEvents(decodedThread.snapshotSequence, 10_000)
9397 .pipe(9428 .pipe(
9398 Stream.runFold(9429 Stream.runFold(
9399 () => decodedThread.snapshotSequence,9430 () => ({
9400 (sequence, event) =>9431 detail: decodedThread.snapshotSequence,
9401 event.aggregateId === TRANSFER_THREAD_ID && isThreadDetailEvent(event)9432 aggregate: decodedShell.snapshotSequence,
9402 ? Math.max(sequence, event.sequence)9433 }),
9403 : sequence,9434 (sequences, event) =>
9435 event.aggregateId !== TRANSFER_THREAD_ID
9436 ? sequences
9437 : {
9438 detail: isThreadDetailEvent(event)
9439 ? Math.max(sequences.detail, event.sequence)
9440 : sequences.detail,
9441 aggregate: Math.max(sequences.aggregate, event.sequence),
9442 },
9404 ),9443 ),
9405 );9444 );
9445 const finalThreadSequence = finalSequences.detail;
9406 assert.isAbove(finalThreadSequence, decodedThread.snapshotSequence);9446 assert.isAbove(finalThreadSequence, decodedThread.snapshotSequence);
94079447
9448 const reachedFinalThreadEvent = (item: OrchestrationThreadStreamItem) =>
9449 item.kind === "event" && item.event.sequence === finalThreadSequence;
9450 // Shell items carry the sequence of the latest coalesced
9451 // event for the thread, so the last one lands at or past
9452 // the final thread event.
9453 const reachedFinalShellEvent = (item: OrchestrationShellStreamItem) =>
9454 item.kind === "thread-upserted" && item.sequence >= finalSequences.aggregate;
9408 yield* collectQueueUntil(9455 yield* collectQueueUntil(
9409 threadItems,9456 threadItems,
9410 (item) =>9457 reachedFinalThreadEvent,
9411 item.kind === "event" && item.event.sequence === finalThreadSequence,
9412 `${provider} thread stream to reach sequence ${finalThreadSequence}`,9458 `${provider} thread stream to reach sequence ${finalThreadSequence}`,
9413 );9459 );
9414 const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals());9460 yield* collectQueueUntil(
9461 secondThreadItems,
9462 reachedFinalThreadEvent,
9463 `${provider} second client thread stream to reach sequence ${finalThreadSequence}`,
9464 );
9465 yield* collectQueueUntil(
9466 shellItems,
9467 reachedFinalShellEvent,
9468 `${provider} shell stream to reach sequence ${finalSequences.aggregate}`,
9469 );
9470 yield* collectQueueUntil(
9471 secondShellItems,
9472 reachedFinalShellEvent,
9473 `${provider} second client shell stream to reach sequence ${finalSequences.aggregate}`,
9474 );
9475 const measuredTurnWebSocket = transferDelta(
9476 turnStartTotals,
9477 threadClient.recorder.totals(),
9478 );
9479 const measuredTurnShellWebSocket = transferDelta(
9480 shellTurnStartTotals,
9481 shellClient.recorder.totals(),
9482 );
9483 const measuredTurnSecondClientWebSocket = transferDelta(
9484 secondTurnStartTotals,
9485 secondClient.recorder.totals(),
9486 );
9487 const measuredTurnSqlStatements = sqlCounter.count() - turnStartSqlStatements;
9488
9489 // The second device drops and comes back with the cursors it
9490 // held before the turn, one subscription at a time so the
9491 // catch-up bytes stay separable.
9492 yield* secondClient.close;
9493 const reconnectSqlStart = sqlCounter.count();
9494 const reconnected = yield* openMeasuredWsClient({ url: wsUrl, cookie });
9495 const reconnectStartTotals = reconnected.recorder.totals();
9496 const reconnectThreadItems = yield* subscribeThreadItems(
9497 reconnected,
9498 decodedThread.snapshotSequence,
9499 );
9500 const reconnectThreadMode = yield* awaitSubscriptionSynchronized(
9501 reconnectThreadItems,
9502 `${provider} reconnected thread subscription to synchronize`,
9503 );
9504 const reconnectThreadTotals = reconnected.recorder.totals();
9505 const reconnectShellItems = yield* subscribeShellItems(
9506 reconnected,
9507 decodedShell.snapshotSequence,
9508 );
9509 const reconnectShellMode = yield* awaitSubscriptionSynchronized(
9510 reconnectShellItems,
9511 `${provider} reconnected shell subscription to synchronize`,
9512 );
9513 const reconnectShellTotals = reconnected.recorder.totals();
9514 const reconnectSqlStatements = sqlCounter.count() - reconnectSqlStart;
94159515
9416 const finalThreadSnapshot = yield* harness.snapshotQuery9516 const finalThreadSnapshot = yield* harness.snapshotQuery
9417 .getThreadDetailSnapshot(TRANSFER_THREAD_ID)9517 .getThreadDetailSnapshot(TRANSFER_THREAD_ID)
...@@ -9436,12 +9536,29 @@ it.live(...@@ -9436,12 +9536,29 @@ it.live(
9436 provider,9536 provider,
9437 threadSnapshot,9537 threadSnapshot,
9438 measuredTurnWebSocket,9538 measuredTurnWebSocket,
9539 shellSnapshot,
9540 measuredTurnShellWebSocket,
9541 measuredTurnSecondClientWebSocket,
9542 reconnectThread: {
9543 mode: reconnectThreadMode,
9544 ...transferDelta(reconnectStartTotals, reconnectThreadTotals),
9545 },
9546 reconnectShell: {
9547 mode: reconnectShellMode,
9548 ...transferDelta(reconnectThreadTotals, reconnectShellTotals),
9549 },
9550 measuredTurnSqlStatements,
9551 reconnectSqlStatements,
9439 } satisfies TransferBudgetRun;9552 } satisfies TransferBudgetRun;
9440 }).pipe(Effect.provide(protocolLayer)),9553 }),
9441 );9554 );
9442 }),9555 }),
9443 (harness) => harness.dispose,9556 (harness) => harness.dispose,
9444 ).pipe(Effect.provide(NodeHttpServerTestWithWsDeflate)),9557 ).pipe(
9558 Effect.provideService(Tracer.Tracer, sqlCounter.tracer),
9559 Effect.provide(NodeHttpServerTestWithWsDeflate),
9560 );
9561 },
9445 { concurrency: 1 },9562 { concurrency: 1 },
9446 );9563 );
94479564