1// @effect-diagnostics nodeBuiltinImport:off - Measures real local HTTP/WebSocket sockets.
2import * as NodeHttp from "node:http";
3import { assert, it } from "@effect/vitest";
4import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
5import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
6import * as NodeServices from "@effect/platform-node/NodeServices";
7import {
8 ORCHESTRATION_PROTOCOL_HEADER,
9 ORCHESTRATION_PROTOCOL_VERSION,
10 AuthSessionId,
11 AuthOrchestrationReadScope,
12 EnvironmentHttpApi,
13 EnvironmentId,
14 EnvironmentAuthenticatedAuth,
15 EnvironmentAuthenticatedPrincipal,
16 ORCHESTRATION_V2_WS_METHODS,
17 OrchestrationV2RpcSchemas,
18 OrchestrationV2ThreadDetailSnapshot,
19 OrchestrationV2ShellSnapshot,
20 OrchestrationV2GetThreadProjectionError,
21 OrchestrationV2GetShellSnapshotError,
22 ProviderDriverKind,
23 type OrchestrationV2ThreadStreamItem,
24 type OrchestrationV2ShellStreamItem,
25} from "@t3tools/contracts";
26import * as Context from "effect/Context";
27import * as Effect from "effect/Effect";
28import * as FileSystem from "effect/FileSystem";
29import * as Layer from "effect/Layer";
30import * as PubSub from "effect/PubSub";
31import * as Queue from "effect/Queue";
32import * as Schema from "effect/Schema";
33import * as Scope from "effect/Scope";
34import * as Stream from "effect/Stream";
35import * as HttpApi from "effect/http-api/HttpApi";
36import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
37import { HttpRouter, HttpServer } from "effect/http";
38import { Rpc, RpcGroup, RpcServer, RpcSerialization } from "effect/rpc";
39import * as ServerEnvironment from "../src/environment/ServerEnvironment.ts";
40import * as SqlitePersistence from "../src/persistence/Sqlite.ts";
41import * as OrchestrationEventStore from "../src/persistence/OrchestrationEventStore.ts";
42import * as EventStore from "../src/orchestration-v2/EventStore.ts";
43import * as EventSink from "../src/orchestration-v2/EventSink.ts";
44import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts";
45import * as ThreadManagementService from "../src/orchestration-v2/ThreadManagementService.ts";
46import * as ProjectStore from "../src/orchestration-v2/ProjectStore.ts";
47import * as ProjectService from "../src/project/ProjectService.ts";
48import * as ProjectEnrichmentService from "../src/project/ProjectEnrichmentService.ts";
49import * as OrchestrationHttp from "../src/orchestration-v2/http.ts";
50import * as ServerHttp from "../src/http.ts";
51import { subscribeOrchestrationV2Thread, subscribeOrchestrationV2Shell } from "../src/ws.ts";
52import {
53 measureHttpGet,
54 openMeasuredWsClient,
55 type MeasuredWsClient,
56} from "./NetworkTransferMeasurement.integration.ts";
57import { makeSqlStatementCounter } from "./SqlStatementCounter.integration.ts";
58import {
59 formatTransferBudgetReport,
60 formatTransferBudgetResult,
61 transferBudgetViolations,
62 type TransferBudgetRun,
63} from "./TransferBudgetReport.integration.ts";
64import { TRANSFER_HISTORY_TURN_COUNT } from "./fixtures/transferBudget.ts";
65import { THREAD_ID, threadCreated, turnEvents } from "./TransferBudgetV2Fixture.integration.ts";
66
67const decodeThreadSnapshot = Schema.decodeUnknownEffect(
68 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot)),
69);
70const decodeShellSnapshot = Schema.decodeUnknownEffect(
71 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellSnapshot)),
72);
73
74const layerPersistence = EventSink.layer.pipe(
75 Layer.provideMerge(EventStore.layerFromOrchestrationEventStore),
76 Layer.provideMerge(ProjectionStore.layer),
77 Layer.provideMerge(OrchestrationEventStore.layer),
78 Layer.provideMerge(SqlitePersistence.layerMemory),
79);
80// The facade's import/command paths are outside this measurement. Its reads and
81// stream delegate to the real SQL projection and event sink, as the runtime does.
82const layerManagement = Layer.unwrap(
83 Effect.gen(function* () {
84 const projections = yield* ProjectionStore.ProjectionStoreV2;
85 const sink = yield* EventSink.EventSinkV2;
86 return Layer.mock(ThreadManagementService.ThreadManagementService)({
87 ensureLegacyTranscript: () => Effect.void,
88 getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
89 getThreadSnapshotWindow: (id, options) =>
90 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
91 getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie),
92 getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie),
93 readShellSnapshot: (options) =>
94 projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie),
95 streamStoredEventsFrom: (input) =>
96 sink.stream({ ...input, bounded: true }).pipe(Stream.orDie),
97 });
98 }),
99);
100const layerEnrichment = Layer.unwrap(
101 Effect.gen(function* () {
102 const changes = yield* PubSub.unbounded<never>();
103 return Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({
104 getAvailable: () =>
105 Effect.succeed({
106 repositoryIdentity: null,
107 faviconPath: null,
108 repositoryIdentityResolved: true,
109 }),
110 subscribeChanges: PubSub.subscribe(changes),
111 });
112 }),
113);
114// The transfer history has no project events, so shell streams never read a project shell.
115const layerServices = layerManagement.pipe(
116 Layer.provideMerge(
117 Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
118 getEnvironmentId: Effect.succeed(EnvironmentId.make("transfer-test")),
119 }),
120 ),
121 Layer.provideMerge(ProjectStore.layer),
122 Layer.provideMerge(Layer.mock(ProjectService.ProjectService)({})),
123 Layer.provideMerge(layerEnrichment),
124 Layer.provideMerge(layerPersistence),
125);
126class TransferApi extends HttpApi.make("environment").add(
127 EnvironmentHttpApi.groups.orchestration,
128) {}
129const layerAuth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) =>
130 effect.pipe(
131 Effect.provideService(EnvironmentAuthenticatedPrincipal, {
132 sessionId: AuthSessionId.make("transfer-session"),
133 subject: "transfer-test",
134 method: "browser-session-cookie",
135 scopes: new Set([AuthOrchestrationReadScope]),
136 }),
137 ),
138);
139const group = RpcGroup.make(
140 Rpc.make(ORCHESTRATION_V2_WS_METHODS.subscribeThread, {
141 payload: OrchestrationV2RpcSchemas.subscribeThread.input,
142 success: OrchestrationV2RpcSchemas.subscribeThread.output,
143 error: OrchestrationV2GetThreadProjectionError,
144 stream: true,
145 }),
146 Rpc.make(ORCHESTRATION_V2_WS_METHODS.subscribeShell, {
147 payload: OrchestrationV2RpcSchemas.subscribeShell.input,
148 success: OrchestrationV2RpcSchemas.subscribeShell.output,
149 error: OrchestrationV2GetShellSnapshotError,
150 stream: true,
151 }),
152);
153const layerHandlers = group.toLayer({
154 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input) =>
155 Stream.unwrap(subscribeOrchestrationV2Thread(input)),
156 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input) =>
157 Stream.unwrap(subscribeOrchestrationV2Shell(input)),
158});
159const layerRoutes = Layer.mergeAll(
160 HttpApiBuilder.layer(TransferApi).pipe(
161 Layer.provide(OrchestrationHttp.layer),
162 Layer.provide(layerAuth),
163 ),
164 RpcServer.layerHttp({ group, path: "/ws", protocol: "websocket" }).pipe(
165 Layer.provide(layerHandlers),
166 Layer.provide(RpcSerialization.layerJson),
167 ),
168).pipe(Layer.provide(ServerHttp.layerHttpCompression), Layer.provide(NodeHttpPlatform.layer));
169
170function difference(
171 after: { wireBytes: number; decodedBytes: number; messages: number },
172 before: typeof after,
173) {
174 return {
175 wireBytes: after.wireBytes - before.wireBytes,
176 decodedBytes: after.decodedBytes - before.decodedBytes,
177 messages: after.messages - before.messages,
178 };
179}
180type StreamItem = OrchestrationV2ThreadStreamItem | OrchestrationV2ShellStreamItem;
181const collectUntil = Effect.fn("transfer.collectUntil")(function* <T extends StreamItem>(
182 queue: Queue.Queue<T>,
183 done: (item: T) => boolean,
184) {
185 const items: T[] = [];
186 while (true) {
187 const item = yield* Queue.take(queue);
188 items.push(item);
189 if (done(item)) return items;
190 }
191});
192const subscribe = Effect.fn("transfer.subscribe")(function* (
193 client: MeasuredWsClient,
194 kind: "thread" | "shell",
195 afterSequence: number,
196) {
197 const queue = yield* Queue.unbounded<StreamItem>();
198 const stream: Stream.Stream<StreamItem, unknown> =
199 kind === "thread"
200 ? client.client[ORCHESTRATION_V2_WS_METHODS.subscribeThread]({
201 threadId: THREAD_ID,
202 afterSequence,
203 requestCompletionMarker: true,
204 acceptBoundedSnapshot: true,
205 })
206 : client.client[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({
207 afterSequence,
208 requestCompletionMarker: true,
209 });
210 yield* stream.pipe(
211 Stream.runForEach((item) => Queue.offer(queue, item)),
212 Scope.provide(client.scope),
213 Effect.forkIn(client.scope),
214 );
215 return queue;
216});
217const synchronized = (queue: Queue.Queue<StreamItem>) =>
218 collectUntil(queue, (item) => item.kind === "synchronized").pipe(
219 Effect.map((items) =>
220 items.some((item) => item.kind === "snapshot") ? ("snapshot" as const) : ("replay" as const),
221 ),
222 );
223
224it.live(
225 "reports V2 thread HTTP and WebSocket transfer budgets",
226 () =>
227 Effect.gen(function* () {
228 const runs: TransferBudgetRun[] = [];
229 for (const provider of [
230 ProviderDriverKind.make("codex"),
231 ProviderDriverKind.make("claudeAgent"),
232 ]) {
233 const counter = makeSqlStatementCounter();
234 const run = yield* Effect.scoped(
235 Effect.gen(function* () {
236 const sink = yield* EventSink.EventSinkV2;
237 const projection = yield* ProjectionStore.ProjectionStoreV2;
238 yield* sink.write({ events: [threadCreated(provider)] });
239 for (let index = 0; index < TRANSFER_HISTORY_TURN_COUNT; index++)
240 yield* sink.write({ events: turnEvents(provider, index, false) });
241 const context = yield* Effect.context<Layer.Success<typeof layerServices>>();
242 const server = yield* Layer.build(
243 HttpRouter.serve(layerRoutes, { disableListenLog: true }).pipe(
244 Layer.provide(Layer.succeedContext(context)),
245 Layer.provideMerge(
246 NodeHttpServer.layer(NodeHttp.createServer, {
247 host: "127.0.0.1",
248 port: 0,
249 websocket: { perMessageDeflate: true },
250 }),
251 ),
252 ),
253 );
254 const address = Context.get(server, HttpServer.HttpServer).address;
255 if (!("port" in address)) return yield* Effect.die("Expected TCP server");
256 const base = `http://127.0.0.1:${address.port}`;
257 const threadSnapshot = yield* measureHttpGet({
258 url: `${base}/api/orchestration/threads/${THREAD_ID}`,
259 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) },
260 });
261 assert.equal(
262 threadSnapshot.status,
263 200,
264 Buffer.from(threadSnapshot.decodedBody).toString(),
265 );
266 assert.equal(threadSnapshot.contentEncoding, "gzip");
267 const decodedThread = yield* decodeThreadSnapshot(
268 Buffer.from(threadSnapshot.decodedBody).toString(),
269 );
270 assert.equal(decodedThread.projection.messages.length, TRANSFER_HISTORY_TURN_COUNT * 2);
271 assert.equal(
272 decodedThread.projection.turnItems.length,
273 TRANSFER_HISTORY_TURN_COUNT * 6,
274 );
275 assert.notInclude(Buffer.from(threadSnapshot.decodedBody).toString(), "digest=");
276 const shellSnapshot = yield* measureHttpGet({
277 url: `${base}/api/orchestration/shell`,
278 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) },
279 });
280 assert.equal(shellSnapshot.status, 200);
281 const decodedShell = yield* decodeShellSnapshot(
282 Buffer.from(shellSnapshot.decodedBody).toString(),
283 );
284 assert.equal(decodedShell.threads.length, 1);
285 const sequence = decodedThread.snapshotSequence;
286 assert.equal(decodedShell.snapshotSequence, sequence);
287 const url = `${base.replace("http:", "ws:")}/ws`;
288 const thread = yield* openMeasuredWsClient({ url, cookie: "" });
289 const shell = yield* openMeasuredWsClient({ url, cookie: "" });
290 const second = yield* openMeasuredWsClient({ url, cookie: "" });
291 assert.include(thread.recorder.negotiatedExtensions(), "permessage-deflate");
292 const tq = yield* subscribe(thread, "thread", sequence);
293 const sq = yield* subscribe(shell, "shell", sequence);
294 const ttq = yield* subscribe(second, "thread", sequence);
295 const ssq = yield* subscribe(second, "shell", sequence);
296 for (const queue of [tq, sq, ttq, ssq])
297 assert.equal(yield* synchronized(queue), "replay");
298 const threadBefore = thread.recorder.totals();
299 const shellBefore = shell.recorder.totals();
300 const secondBefore = second.recorder.totals();
301 const sqlBefore = counter.count();
302 const events = turnEvents(provider, TRANSFER_HISTORY_TURN_COUNT, true);
303 const stored = yield* sink.write({ events });
304 const last = stored.at(-1)!;
305 const terminal = (item: StreamItem) =>
306 "sequence" in item && item.sequence >= last.sequence;
307 for (const queue of [tq, sq, ttq, ssq]) yield* collectUntil(queue, terminal);
308 const measuredTurnWebSocket = difference(thread.recorder.totals(), threadBefore);
309 const measuredTurnShellWebSocket = difference(shell.recorder.totals(), shellBefore);
310 const measuredTurnSecondClientWebSocket = difference(
311 second.recorder.totals(),
312 secondBefore,
313 );
314 const measuredTurnSqlStatements = counter.count() - sqlBefore;
315 // Verify large output is retained durably even though the transport omits it.
316 const durable = yield* projection.getThreadProjection(THREAD_ID);
317 assert.isTrue(
318 durable.turnItems.some(
319 (item) =>
320 item.type === "dynamic_tool" && JSON.stringify(item.output).length > 1_000_000,
321 ),
322 );
323 yield* second.close;
324 const reconnectSqlStart = counter.count();
325 const resumedThread = yield* openMeasuredWsClient({ url, cookie: "" });
326 const resumedShell = yield* openMeasuredWsClient({ url, cookie: "" });
327 const rq = yield* subscribe(resumedThread, "thread", sequence);
328 const rsq = yield* subscribe(resumedShell, "shell", sequence);
329 const threadMode = yield* synchronized(rq);
330 const shellMode = yield* synchronized(rsq);
331 const reconnectThread = { ...resumedThread.recorder.totals(), mode: threadMode };
332 const reconnectShell = { ...resumedShell.recorder.totals(), mode: shellMode };
333 assert.equal(threadMode, "snapshot"); // retained MCP payload exceeds raw replay budget
334 return {
335 provider,
336 threadSnapshot,
337 shellSnapshot,
338 measuredTurnWebSocket,
339 measuredTurnShellWebSocket,
340 measuredTurnSecondClientWebSocket,
341 measuredTurnSqlStatements,
342 reconnectThread,
343 reconnectShell,
344 reconnectSqlStatements: counter.count() - reconnectSqlStart,
345 } satisfies TransferBudgetRun;
346 }).pipe(
347 Effect.provide(Layer.fresh(layerServices).pipe(Layer.provideMerge(NodeServices.layer))),
348 Effect.withTracer(counter.tracer),
349 ),
350 );
351 runs.push(run);
352 }
353 const fs = yield* FileSystem.FileSystem;
354 const report = formatTransferBudgetReport(runs);
355 for (const [path, contents] of [
356 [process.env.T3CODE_TRANSFER_BUDGET_REPORT_PATH, report],
357 [process.env.T3CODE_TRANSFER_BUDGET_RESULT_PATH, formatTransferBudgetResult(runs)],
358 ])
359 if (path && contents) yield* fs.writeFileString(path, contents);
360 yield* Effect.log(report);
361 assert.deepEqual(transferBudgetViolations(runs), []);
362 }).pipe(Effect.provide(NodeServices.layer)),
363 30_000,
364);