1// @effect-diagnostics nodeBuiltinImport:off - Measures real local HTTP/WebSocket sockets.
2import * as NodeHttp from "node:http";
3import * as NodeUtil from "node:util";
4import { assert, it } from "@effect/vitest";
5import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
6import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
7import * as NodeServices from "@effect/platform-node/NodeServices";
8import {
9 ORCHESTRATION_PROTOCOL_HEADER,
10 ORCHESTRATION_PROTOCOL_VERSION,
11 AuthSessionId,
12 AuthOrchestrationReadScope,
13 EnvironmentHttpApi,
14 EnvironmentId,
15 EnvironmentAuthenticatedAuth,
16 EnvironmentAuthenticatedPrincipal,
17 ORCHESTRATION_V2_WS_METHODS,
18 OrchestrationV2RpcSchemas,
19 OrchestrationV2ThreadBoundedSnapshot,
20 OrchestrationV2ThreadDetailSnapshot,
21 OrchestrationV2ThreadHistoryPage,
22 OrchestrationV2ShellHttpSnapshot,
23 OrchestrationV2ShellResumable,
24 OrchestrationV2GetThreadProjectionError,
25 OrchestrationV2GetShellSnapshotError,
26 ProviderDriverKind,
27 type EnvironmentOrchestrationShellSnapshotQuery,
28 type EnvironmentOrchestrationThreadBoundedSnapshotQuery,
29 type EnvironmentOrchestrationThreadHistoryQuery,
30 type OrchestrationV2SubscribeShellInput,
31 type OrchestrationV2SubscribeThreadInput,
32 type OrchestrationV2ThreadStreamItem,
33 type OrchestrationV2ShellStreamItem,
34 type ThreadId,
35} from "@t3tools/contracts";
36import * as Context from "effect/Context";
37import * as Config from "effect/Config";
38import * as Effect from "effect/Effect";
39import * as FileSystem from "effect/FileSystem";
40import * as Layer from "effect/Layer";
41import * as PubSub from "effect/PubSub";
42import * as Queue from "effect/Queue";
43import * as Schema from "effect/Schema";
44import * as Scope from "effect/Scope";
45import * as Stream from "effect/Stream";
46import * as HttpApi from "effect/http-api/HttpApi";
47import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
48import { HttpRouter, HttpServer } from "effect/http";
49import { Rpc, RpcGroup, RpcServer, RpcSerialization } from "effect/rpc";
50import * as ServerEnvironment from "../src/environment/ServerEnvironment.ts";
51import * as SqlitePersistence from "../src/persistence/Sqlite.ts";
52import * as OrchestrationEventStore from "../src/persistence/OrchestrationEventStore.ts";
53import * as EventStore from "../src/orchestration-v2/EventStore.ts";
54import * as EventSink from "../src/orchestration-v2/EventSink.ts";
55import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts";
56import * as ThreadManagementService from "../src/orchestration-v2/ThreadManagementService.ts";
57import * as ProjectStore from "../src/orchestration-v2/ProjectStore.ts";
58import * as ProjectService from "../src/project/ProjectService.ts";
59import * as ProjectEnrichmentService from "../src/project/ProjectEnrichmentService.ts";
60import * as OrchestrationHttp from "../src/orchestration-v2/http.ts";
61import * as ServerHttp from "../src/http.ts";
62import { subscribeOrchestrationV2Thread, subscribeOrchestrationV2Shell } from "../src/ws.ts";
63import {
64 measureHttpGet,
65 openMeasuredWsClient,
66 type HttpTransferMeasurement,
67 type MeasuredWsClient,
68} from "./NetworkTransferMeasurement.integration.ts";
69import { makeSqlStatementCounter } from "./SqlStatementCounter.integration.ts";
70import {
71 formatTransferBudgetReport,
72 formatTransferBudgetResult,
73 formatTransferBudgetScenarios,
74 transferBudgetViolations,
75 type TransferBudgetRun,
76} from "./TransferBudgetReport.integration.ts";
77import { TRANSFER_HISTORY_TURN_COUNT } from "./fixtures/transferBudget.ts";
78import {
79 FLEET_PROJECT_IDENTITY,
80 fleetProjectCreated,
81 LONG_TURN_THREAD_ID,
82 longTurnThreadBatches,
83 SHELL_IDLE_SUBAGENTS,
84 SHELL_SETTLED_THREADS,
85 shellFleetEvents,
86 STALE_TODO_LAST_FINISHED_CHECKPOINT_ID,
87 STALE_TODO_LIVE_RUN_ID,
88 STALE_TODO_SUBAGENTS_PER_RUN,
89 STALE_TODO_THREAD_ID,
90 staleTodoListThreadBatches,
91 THREAD_ID,
92 threadCreated,
93 turnEvents,
94} from "./TransferBudgetV2Fixture.integration.ts";
95
96// The opt-ins current web and mobile clients send, as client-runtime's boundedThreadSnapshotHttp.ts,
97// threadHistoryHttp.ts, shellSnapshotHttp.ts, threads.ts and shell.ts send them. Each is `Required`,
98// so an opt-in added to a contract fails typecheck here until this test sends it. An omitted field
99// differs per request (a cursor, or what the client holds), so each scenario sends its own.
100const CLIENT_BOUNDED_OPT_INS = {
101 compactTurnItems: "1",
102 dedupeBodies: "1",
103 compactIds: "1",
104 deferWork: "1",
105} satisfies Required<typeof EnvironmentOrchestrationThreadBoundedSnapshotQuery.Type>;
106const CLIENT_HISTORY_OPT_INS = { compactIds: "1", deferWork: "1" } satisfies Required<
107 Omit<typeof EnvironmentOrchestrationThreadHistoryQuery.Type, "cursor" | "throughEntryId" | "view">
108>;
109const CLIENT_SHELL_OPT_INS = {
110 omitIdleSubagents: "1",
111 omitOlderSettled: "1",
112} satisfies Required<Omit<typeof EnvironmentOrchestrationShellSnapshotQuery.Type, "resumeFrom">>;
113const CLIENT_THREAD_SUBSCRIPTION = {
114 requestCompletionMarker: true,
115 acceptBoundedSnapshot: true,
116 acceptDeferredWork: true,
117 acceptCompactTurnItems: true,
118 acceptCompactIds: true,
119 acceptDedupedBodies: true,
120} satisfies Required<Omit<OrchestrationV2SubscribeThreadInput, "threadId" | "afterSequence">>;
121type ShellResumeField = "afterSequence" | "httpSnapshotFallback" | "knownProjectIdentities";
122const CLIENT_SHELL_SUBSCRIPTION = {
123 requestCompletionMarker: true,
124 omitIdleSubagents: true,
125 omitOlderSettled: true,
126} satisfies Required<Omit<OrchestrationV2SubscribeShellInput, ShellResumeField>>;
127const query = (optIns: Readonly<Record<string, string>>) => new URLSearchParams(optIns).toString();
128const CLIENT_BOUNDED_QUERY = query(CLIENT_BOUNDED_OPT_INS);
129const CLIENT_HISTORY_QUERY = query(CLIENT_HISTORY_OPT_INS);
130const CLIENT_SHELL_QUERY = query(CLIENT_SHELL_OPT_INS);
131// A client that holds shell rows names their cursor in the same request.
132const clientShellResumeQuery = (resumeFrom: number) =>
133 query({
134 ...CLIENT_SHELL_OPT_INS,
135 resumeFrom: String(resumeFrom),
136 } satisfies Required<typeof EnvironmentOrchestrationShellSnapshotQuery.Type>);
137
138const decodeThreadSnapshot = Schema.decodeUnknownEffect(
139 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot)),
140);
141const decodeBoundedSnapshot = Schema.decodeUnknownEffect(
142 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)),
143);
144const decodeHistoryPage = Schema.decodeUnknownEffect(
145 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage)),
146);
147const decodeShellSnapshot = Schema.decodeUnknownEffect(
148 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellHttpSnapshot)),
149);
150const decodeShellResumable = Schema.decodeUnknownEffect(
151 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellResumable)),
152);
153
154const layerPersistence = EventSink.layer.pipe(
155 Layer.provideMerge(EventStore.layerFromOrchestrationEventStore),
156 Layer.provideMerge(ProjectionStore.layer),
157 Layer.provideMerge(OrchestrationEventStore.layer),
158 Layer.provideMerge(SqlitePersistence.layerMemory),
159);
160const layerManagement = Layer.unwrap(
161 Effect.gen(function* () {
162 const projections = yield* ProjectionStore.ProjectionStoreV2;
163 const sink = yield* EventSink.EventSinkV2;
164 return Layer.mock(ThreadManagementService.ThreadManagementService)({
165 ensureLegacyTranscript: () => Effect.void,
166 getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
167 getThreadSnapshotWindow: (id, options) =>
168 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
169 getThreadHistoryPage: (id, cursor, throughEntryId, conversationOnly) =>
170 projections
171 .getThreadHistoryPage(id, cursor, throughEntryId, conversationOnly)
172 .pipe(Effect.orDie),
173 getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie),
174 getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie),
175 listActiveShellIndex: () => projections.listActiveShellIndex().pipe(Effect.orDie),
176 listActiveSubagentThreadIds: () =>
177 projections.listActiveSubagentThreadIds().pipe(Effect.orDie),
178 readShellSnapshot: (options) =>
179 projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie),
180 streamStoredEventsFrom: (input) =>
181 sink.stream({ ...input, bounded: true }).pipe(Stream.orDie),
182 });
183 }),
184);
185const layerEnrichment = Layer.unwrap(
186 Effect.gen(function* () {
187 const changes = yield* PubSub.unbounded<never>();
188 return Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({
189 getAvailable: () =>
190 Effect.succeed({
191 repositoryIdentity: FLEET_PROJECT_IDENTITY,
192 faviconPath: null,
193 repositoryIdentityResolved: true,
194 }),
195 subscribeChanges: PubSub.subscribe(changes),
196 });
197 }),
198);
199const layerServices = layerManagement.pipe(
200 Layer.provideMerge(
201 Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
202 getEnvironmentId: Effect.succeed(EnvironmentId.make("transfer-test")),
203 }),
204 ),
205 Layer.provideMerge(ProjectStore.layer),
206 Layer.provideMerge(Layer.mock(ProjectService.ProjectService)({})),
207 Layer.provideMerge(layerEnrichment),
208 Layer.provideMerge(layerPersistence),
209);
210class TransferApi extends HttpApi.make("environment").add(
211 EnvironmentHttpApi.groups.orchestration,
212) {}
213const layerAuth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) =>
214 effect.pipe(
215 Effect.provideService(EnvironmentAuthenticatedPrincipal, {
216 sessionId: AuthSessionId.make("transfer-session"),
217 subject: "transfer-test",
218 method: "browser-session-cookie",
219 scopes: new Set([AuthOrchestrationReadScope]),
220 }),
221 ),
222);
223const group = RpcGroup.make(
224 Rpc.make(ORCHESTRATION_V2_WS_METHODS.subscribeThread, {
225 payload: OrchestrationV2RpcSchemas.subscribeThread.input,
226 success: OrchestrationV2RpcSchemas.subscribeThread.output,
227 error: OrchestrationV2GetThreadProjectionError,
228 stream: true,
229 }),
230 Rpc.make(ORCHESTRATION_V2_WS_METHODS.subscribeShell, {
231 payload: OrchestrationV2RpcSchemas.subscribeShell.input,
232 success: OrchestrationV2RpcSchemas.subscribeShell.output,
233 error: OrchestrationV2GetShellSnapshotError,
234 stream: true,
235 }),
236);
237const layerHandlers = group.toLayer({
238 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input) =>
239 Stream.unwrap(subscribeOrchestrationV2Thread(input)),
240 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input) =>
241 Stream.unwrap(subscribeOrchestrationV2Shell(input)),
242});
243const layerRoutes = Layer.mergeAll(
244 HttpApiBuilder.layer(TransferApi).pipe(
245 Layer.provide(OrchestrationHttp.layer),
246 Layer.provide(layerAuth),
247 ),
248 RpcServer.layerHttp({ group, path: "/ws", protocol: "websocket" }).pipe(
249 Layer.provide(layerHandlers),
250 Layer.provide(RpcSerialization.layerJson),
251 ),
252).pipe(Layer.provide(ServerHttp.layerHttpCompression), Layer.provide(NodeHttpPlatform.layer));
253
254function difference(
255 after: { wireBytes: number; decodedBytes: number; messages: number },
256 before: typeof after,
257) {
258 return {
259 wireBytes: after.wireBytes - before.wireBytes,
260 decodedBytes: after.decodedBytes - before.decodedBytes,
261 messages: after.messages - before.messages,
262 };
263}
264type StreamItem = OrchestrationV2ThreadStreamItem | OrchestrationV2ShellStreamItem;
265const collectUntil = Effect.fn("transfer.collectUntil")(function* <T extends StreamItem>(
266 queue: Queue.Queue<T>,
267 done: (item: T) => boolean,
268) {
269 const items: T[] = [];
270 while (true) {
271 const item = yield* Queue.take(queue);
272 items.push(item);
273 if (done(item)) return items;
274 }
275});
276type Subscription =
277 | ({ readonly thread: ThreadId } & Pick<OrchestrationV2SubscribeThreadInput, "afterSequence">)
278 | ({ readonly shell: true } & Pick<OrchestrationV2SubscribeShellInput, ShellResumeField>);
279function subscriptionStream(
280 client: MeasuredWsClient,
281 subscription: Subscription,
282): Stream.Stream<StreamItem, unknown> {
283 if ("thread" in subscription) {
284 const { thread, ...resume } = subscription;
285 return client.client[ORCHESTRATION_V2_WS_METHODS.subscribeThread]({
286 ...CLIENT_THREAD_SUBSCRIPTION,
287 ...resume,
288 threadId: thread,
289 });
290 }
291 const { shell: _, ...resume } = subscription;
292 return client.client[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({
293 ...CLIENT_SHELL_SUBSCRIPTION,
294 ...resume,
295 });
296}
297const subscribe = Effect.fn("transfer.subscribe")(function* (
298 client: MeasuredWsClient,
299 subscription: Subscription,
300) {
301 const queue = yield* Queue.unbounded<StreamItem>();
302 yield* subscriptionStream(client, subscription).pipe(
303 Stream.runForEach((item) => Queue.offer(queue, item)),
304 Scope.provide(client.scope),
305 Effect.forkIn(client.scope),
306 );
307 return queue;
308});
309const synchronized = (queue: Queue.Queue<StreamItem>) =>
310 collectUntil(queue, (item) => item.kind === "synchronized").pipe(
311 Effect.map((items) =>
312 items.some((item) => item.kind === "snapshot") ? ("snapshot" as const) : ("replay" as const),
313 ),
314 );
315
316it.live("reports V2 thread HTTP and WebSocket transfer budgets", () =>
317 Effect.gen(function* () {
318 const runs: TransferBudgetRun[] = [];
319 for (const provider of [
320 ProviderDriverKind.make("codex"),
321 ProviderDriverKind.make("claudeAgent"),
322 ]) {
323 const counter = makeSqlStatementCounter();
324 const run = yield* Effect.scoped(
325 Effect.gen(function* () {
326 const sink = yield* EventSink.EventSinkV2;
327 const projection = yield* ProjectionStore.ProjectionStoreV2;
328 yield* sink.write({ events: [threadCreated(provider)] });
329 for (let index = 0; index < TRANSFER_HISTORY_TURN_COUNT; index++)
330 yield* sink.write({ events: turnEvents(provider, index, false) });
331 const context = yield* Effect.context<Layer.Success<typeof layerServices>>();
332 const server = yield* Layer.build(
333 HttpRouter.serve(layerRoutes, { disableListenLog: true }).pipe(
334 Layer.provide(Layer.succeedContext(context)),
335 Layer.provideMerge(
336 NodeHttpServer.layer(NodeHttp.createServer, {
337 host: "127.0.0.1",
338 port: 0,
339 websocket: { perMessageDeflate: true },
340 }),
341 ),
342 ),
343 );
344 const address = Context.get(server, HttpServer.HttpServer).address;
345 if (!("port" in address)) return yield* Effect.die("Expected TCP server");
346 const base = `http://127.0.0.1:${address.port}`;
347 const threadSnapshot = yield* measureHttpGet({
348 url: `${base}/api/orchestration/threads/${THREAD_ID}`,
349 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) },
350 });
351 assert.equal(
352 threadSnapshot.status,
353 200,
354 Buffer.from(threadSnapshot.decodedBody).toString(),
355 );
356 assert.equal(threadSnapshot.contentEncoding, "gzip");
357 const decodedThread = yield* decodeThreadSnapshot(
358 Buffer.from(threadSnapshot.decodedBody).toString(),
359 );
360 assert.equal(decodedThread.projection.messages.length, TRANSFER_HISTORY_TURN_COUNT * 2);
361 assert.equal(decodedThread.projection.turnItems.length, TRANSFER_HISTORY_TURN_COUNT * 6);
362 assert.notInclude(Buffer.from(threadSnapshot.decodedBody).toString(), "digest=");
363 const shellSnapshot = yield* measureHttpGet({
364 url: `${base}/api/orchestration/shell`,
365 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) },
366 });
367 assert.equal(shellSnapshot.status, 200);
368 const decodedShell = yield* decodeShellSnapshot(
369 Buffer.from(shellSnapshot.decodedBody).toString(),
370 );
371 assert.equal(decodedShell.threads.length, 1);
372 const sequence = decodedThread.snapshotSequence;
373 assert.equal(decodedShell.snapshotSequence, sequence);
374 const url = `${base.replace("http:", "ws:")}/ws`;
375 const thread = yield* openMeasuredWsClient({ url, cookie: "" });
376 const shell = yield* openMeasuredWsClient({ url, cookie: "" });
377 const second = yield* openMeasuredWsClient({ url, cookie: "" });
378 assert.include(thread.recorder.negotiatedExtensions(), "permessage-deflate");
379 const fromSnapshot = { afterSequence: sequence };
380 const tq = yield* subscribe(thread, { thread: THREAD_ID, ...fromSnapshot });
381 const sq = yield* subscribe(shell, { shell: true, ...fromSnapshot });
382 const ttq = yield* subscribe(second, { thread: THREAD_ID, ...fromSnapshot });
383 const ssq = yield* subscribe(second, { shell: true, ...fromSnapshot });
384 for (const queue of [tq, sq, ttq, ssq])
385 assert.equal(yield* synchronized(queue), "replay");
386 const threadBefore = thread.recorder.totals();
387 const shellBefore = shell.recorder.totals();
388 const secondBefore = second.recorder.totals();
389 const sqlBefore = counter.count();
390 const events = turnEvents(provider, TRANSFER_HISTORY_TURN_COUNT, true);
391 const stored = yield* sink.write({ events });
392 const last = stored.at(-1)!;
393 const threadTerminal = (item: StreamItem) =>
394 "sequence" in item && item.sequence >= last.sequence;
395 // Shell items carry the shell read when their batch is projected, and unchanged
396 // shells are skipped, so a turn split across batches may never yield an item at
397 // the last sequence. Wait for the final shell instead.
398 const finalShell = yield* projection.getThreadShell(THREAD_ID);
399 const shellTerminal = (item: StreamItem) =>
400 item.kind === "thread.updated" && NodeUtil.isDeepStrictEqual(item.thread, finalShell);
401 const liveTurn = yield* collectUntil(tq, threadTerminal);
402 yield* collectUntil(ttq, threadTerminal);
403 // A direct check, so live dedupe stays guarded however small the turn gets.
404 const liveAnswer = liveTurn.flatMap((item) =>
405 item.kind === "event" &&
406 item.event.type === "turn-item.updated" &&
407 item.event.payload.type === "assistant_message"
408 ? [item.event.payload]
409 : [],
410 );
411 assert.equal(liveAnswer.length, 1);
412 assert.equal(liveAnswer[0]!.textOmitted, true);
413 for (const queue of [sq, ssq]) yield* collectUntil(queue, shellTerminal);
414 const measuredTurnWebSocket = difference(thread.recorder.totals(), threadBefore);
415 const measuredTurnShellWebSocket = difference(shell.recorder.totals(), shellBefore);
416 const measuredTurnSecondClientWebSocket = difference(
417 second.recorder.totals(),
418 secondBefore,
419 );
420 const measuredTurnSqlStatements = counter.count() - sqlBefore;
421 const durable = yield* projection.getThreadProjection(THREAD_ID);
422 assert.isTrue(
423 durable.turnItems.some(
424 (item) =>
425 item.type === "dynamic_tool" && JSON.stringify(item.output).length > 1_000_000,
426 ),
427 );
428 yield* second.close;
429 const reconnectSqlStart = counter.count();
430 const resumedThread = yield* openMeasuredWsClient({ url, cookie: "" });
431 const resumedShell = yield* openMeasuredWsClient({ url, cookie: "" });
432 const rq = yield* subscribe(resumedThread, { thread: THREAD_ID, ...fromSnapshot });
433 const rsq = yield* subscribe(resumedShell, { shell: true, ...fromSnapshot });
434 const threadMode = yield* synchronized(rq);
435 const shellMode = yield* synchronized(rsq);
436 const reconnectThread = { ...resumedThread.recorder.totals(), mode: threadMode };
437 const reconnectShell = { ...resumedShell.recorder.totals(), mode: shellMode };
438 const reconnectSqlStatements = counter.count() - reconnectSqlStart;
439 assert.equal(threadMode, "snapshot");
440
441 const get = (path: string) =>
442 measureHttpGet({
443 url: `${base}${path}`,
444 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) },
445 }).pipe(Effect.tap((http) => Effect.sync(() => assert.equal(http.status, 200))));
446 const decoded = (http: HttpTransferMeasurement) =>
447 Buffer.from(http.decodedBody).toString();
448 const longTurnPath = `/api/orchestration/threads/${LONG_TURN_THREAD_ID}`;
449 const readColdOpen = (query: string) =>
450 Effect.gen(function* () {
451 const http = yield* get(`${longTurnPath}/bounded?${query}`);
452 const snapshot = yield* decodeBoundedSnapshot(decoded(http));
453 const rows = snapshot.projection.visibleTurnItems;
454 const workAnchor = rows.find((row) => row.item.deferredWork !== undefined);
455 return { http, rows: rows.length, cursor: snapshot.historyCursor, workAnchor };
456 });
457 const readHistoryPage = (cursor: string, query: string) =>
458 Effect.gen(function* () {
459 const http = yield* get(
460 `${longTurnPath}/history?cursor=${encodeURIComponent(cursor)}&${query}`,
461 );
462 const page = yield* decodeHistoryPage(decoded(http));
463 assert.isTrue(page.hasMoreHistory);
464 return { http, rows: page.items.length };
465 });
466 const readShell = (query: string) =>
467 Effect.gen(function* () {
468 const http = yield* get(`/api/orchestration/shell?${query}`);
469 const { threads, snapshotSequence, projectIdentities } = yield* decodeShellSnapshot(
470 decoded(http),
471 );
472 return { http, rows: threads.length, snapshotSequence, projectIdentities };
473 });
474 // A new socket, read until the subscription caught up or the server ended it.
475 const open = (subscription: Subscription) =>
476 Effect.gen(function* () {
477 const socket = yield* openMeasuredWsClient({ url, cookie: "" });
478 const items = yield* subscriptionStream(socket, subscription).pipe(
479 Stream.takeUntil((item) => item.kind === "synchronized"),
480 Stream.runCollect,
481 Scope.provide(socket.scope),
482 Effect.orDie,
483 );
484 const totals = socket.recorder.totals();
485 yield* socket.close;
486 return { kinds: items.map((item) => item.kind), items, totals };
487 });
488
489 // A phone that comes back after one ordinary turn replays it, without tool bodies.
490 const shortTurn = yield* sink.write({
491 events: turnEvents(provider, TRANSFER_HISTORY_TURN_COUNT + 1, false),
492 });
493 const shortTurnResume = yield* open({ thread: THREAD_ID, afterSequence: last.sequence });
494 assert.notInclude(shortTurnResume.kinds, "snapshot");
495 // The byte cap below also passes when the resume skips the turn.
496 assert.isTrue(
497 shortTurnResume.items.some(
498 (item) => "sequence" in item && item.sequence === shortTurn.at(-1)!.sequence,
499 ),
500 );
501
502 // A turn that is still running cannot defer its work, so it loads in row-budget windows.
503 const [finishLongTurn, ...longTurnHistory] = longTurnThreadBatches(provider).toReversed();
504 for (const events of longTurnHistory.toReversed()) yield* sink.write({ events });
505 const longTurnRunningColdOpenLegacy = yield* readColdOpen("compactTurnItems=1");
506 const longTurnRunningColdOpen = yield* readColdOpen(CLIENT_BOUNDED_QUERY);
507 const { cursor } = longTurnRunningColdOpen;
508 if (cursor === null) return yield* Effect.die("Expected older long-turn history");
509 const longTurnRunningHistoryPageLegacy = yield* readHistoryPage(cursor, "");
510 const longTurnRunningHistoryPage = yield* readHistoryPage(cursor, CLIENT_HISTORY_QUERY);
511
512 // Once it finishes, its work stays on the server until a group is expanded.
513 yield* sink.write({ events: finishLongTurn! });
514 const longTurnColdOpen = yield* readColdOpen(CLIENT_BOUNDED_QUERY);
515 const { workAnchor } = longTurnColdOpen;
516 if (workAnchor?.item.deferredWork === undefined)
517 return yield* Effect.die("Expected the finished long turn to defer its work");
518 const longTurnWorkPage = yield* readHistoryPage(
519 workAnchor.item.deferredWork.cursor,
520 `view=work&throughEntryId=${encodeURIComponent(workAnchor.sourceItemId)}&compactIds=1`,
521 );
522 // Mobile opens a thread over the socket alone.
523 const socketOpen = yield* open({ thread: LONG_TURN_THREAD_ID });
524 const socketSnapshot = socketOpen.items.find(
525 (item) => item.kind === "snapshot" && "projection" in item,
526 );
527 if (socketSnapshot?.kind !== "snapshot" || !("projection" in socketSnapshot))
528 return yield* Effect.die("Expected a bounded socket snapshot");
529 const longTurnSocketColdOpen = {
530 ...socketOpen,
531 rows: socketSnapshot.projection.visibleTurnItems.length,
532 };
533 // The row cap below also passes when the socket ignores the deferral and sends a window.
534 assert.equal(longTurnSocketColdOpen.rows, longTurnColdOpen.rows);
535 const idleThreadResume = yield* open({
536 thread: LONG_TURN_THREAD_ID,
537 afterSequence: socketSnapshot.snapshotSequence,
538 });
539 assert.deepEqual(idleThreadResume.kinds, ["synchronized"]);
540
541 // The transfer thread plus the long-turn thread.
542 const earlierThreads = decodedShell.threads.length + 1;
543 // The fleet's project is in a repository. Earlier phases have no project, as on `main`.
544 yield* (yield* ProjectStore.ProjectStoreV2).apply(fleetProjectCreated);
545 yield* sink.write({ events: shellFleetEvents(provider) });
546 const fleetShellLegacy = yield* readShell("");
547 const fleetShell = yield* readShell(CLIENT_SHELL_QUERY);
548 assert.equal(
549 fleetShellLegacy.rows,
550 earlierThreads + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS,
551 );
552 const { projectIdentities } = fleetShell;
553 if (projectIdentities === undefined)
554 return yield* Effect.die("Expected the shell snapshot to name its project identities");
555 // A cold connect subscribes from the snapshot it just loaded and names its identities.
556 // Without them the resume sends the project again, in a marked snapshot.
557 const fleetShellSubscribe = yield* open({
558 shell: true,
559 afterSequence: fleetShell.snapshotSequence,
560 knownProjectIdentities: projectIdentities,
561 });
562 assert.notInclude(fleetShellSubscribe.kinds, "snapshot");
563 // A client that holds those rows asks the route whether they still resume, and keeps them.
564 const fleetShellResumeHttp = yield* get(
565 `/api/orchestration/shell?${clientShellResumeQuery(fleetShell.snapshotSequence)}`,
566 );
567 yield* decodeShellResumable(decoded(fleetShellResumeHttp));
568 const fleetShellResume = { http: fleetShellResumeHttp, rows: 0 };
569 // A cursor too old to replay gets the snapshot in the same answer.
570 const staleShellReload = yield* readShell(clientShellResumeQuery(sequence));
571 assert.equal(staleShellReload.rows, fleetShell.rows);
572 // When that request fails, the socket tells the client to reload, with no snapshot.
573 const staleShellResume = yield* open({
574 shell: true,
575 afterSequence: sequence,
576 httpSnapshotFallback: true,
577 });
578 assert.deepEqual(staleShellResume.kinds, ["snapshot-required"]);
579
580 for (const events of staleTodoListThreadBatches(provider)) yield* sink.write({ events });
581 const staleTodoHttp = yield* get(
582 `/api/orchestration/threads/${STALE_TODO_THREAD_ID}/bounded?${CLIENT_BOUNDED_QUERY}`,
583 );
584 const staleTodo = (yield* decodeBoundedSnapshot(decoded(staleTodoHttp))).projection;
585 // Clients show the diff of the last finished turn from this record.
586 assert.include(
587 staleTodo.checkpoints.map((checkpoint) => checkpoint.id),
588 STALE_TODO_LAST_FINISHED_CHECKPOINT_ID,
589 );
590 // The caps below also pass when the running turn loses its own tasks.
591 assert.lengthOf(
592 staleTodo.subagents.filter((subagent) => subagent.runId === STALE_TODO_LIVE_RUN_ID),
593 STALE_TODO_SUBAGENTS_PER_RUN,
594 );
595 // The byte cap below also passes when the window loses the handoff.
596 assert.lengthOf(staleTodo.contextHandoffs, 1);
597 const staleTodoColdOpen = {
598 http: staleTodoHttp,
599 rows: staleTodo.visibleTurnItems.length,
600 records: {
601 nodes: staleTodo.nodes.length,
602 subagents: staleTodo.subagents.length,
603 plans: staleTodo.plans.length,
604 providerTurns: staleTodo.providerTurns.length,
605 checkpoints: staleTodo.checkpoints.length,
606 },
607 };
608 return {
609 provider,
610 threadSnapshot,
611 shellSnapshot,
612 measuredTurnWebSocket,
613 measuredTurnShellWebSocket,
614 measuredTurnSecondClientWebSocket,
615 measuredTurnSqlStatements,
616 reconnectThread,
617 reconnectShell,
618 reconnectSqlStatements,
619 shortTurnResume,
620 longTurnRunningColdOpenLegacy,
621 longTurnRunningColdOpen,
622 longTurnRunningHistoryPageLegacy,
623 longTurnRunningHistoryPage,
624 longTurnColdOpen,
625 longTurnWorkPage,
626 longTurnSocketColdOpen,
627 idleThreadResume,
628 fleetShellLegacy,
629 fleetShell,
630 fleetShellSubscribe,
631 fleetShellResume,
632 staleShellReload,
633 staleShellResume,
634 staleTodoColdOpen,
635 } satisfies TransferBudgetRun;
636 }).pipe(
637 Effect.provide(Layer.fresh(layerServices).pipe(Layer.provideMerge(NodeServices.layer))),
638 Effect.withTracer(counter.tracer),
639 ),
640 );
641 runs.push(run);
642 }
643 const fs = yield* FileSystem.FileSystem;
644 const report = formatTransferBudgetReport(runs);
645 const outputPaths = yield* Config.all({
646 report: Config.String("T3CODE_TRANSFER_BUDGET_REPORT_PATH").pipe(Config.withDefault("")),
647 result: Config.String("T3CODE_TRANSFER_BUDGET_RESULT_PATH").pipe(Config.withDefault("")),
648 scenarios: Config.String("T3CODE_TRANSFER_BUDGET_SCENARIOS_PATH").pipe(
649 Config.withDefault(""),
650 ),
651 });
652 for (const [path, contents] of [
653 [outputPaths.report, report],
654 [outputPaths.result, formatTransferBudgetResult(runs)],
655 [outputPaths.scenarios, formatTransferBudgetScenarios(runs)],
656 ])
657 if (path && contents) yield* fs.writeFileString(path, contents);
658 yield* Effect.log(report);
659 assert.deepEqual(transferBudgetViolations(runs), []);
660 }).pipe(Effect.provide(NodeServices.layer)),
661);