1import {
2 CommandId,
3 EnvironmentId,
4 MessageId,
5 ORCHESTRATION_V2_WS_METHODS,
6 OrchestrationV2RpcSchemas,
7 TurnItemId,
8 type OrchestrationV2Command,
9 type OrchestrationV2ProjectedTurnItem,
10 type OrchestrationV2ThreadProjection,
11 type OrchestrationV2TurnItem,
12} from "@t3tools/contracts";
13import { describe, expect, it } from "@effect/vitest";
14import * as Crypto from "effect/Crypto";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Schema from "effect/Schema";
19import * as SubscriptionRef from "effect/SubscriptionRef";
20import type * as RpcMessage from "effect/rpc/RpcMessage";
21
22import {
23 AVAILABLE_CONNECTION_STATE,
24 PrimaryConnectionTarget,
25 type PreparedConnection,
26} from "../connection/model.ts";
27import * as EnvironmentSupervisor from "../connection/supervisor.ts";
28import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
29import * as RpcSession from "../rpc/session.ts";
30import { v2Now, v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts";
31import { startThreadTurn } from "./commands.ts";
32
33const ROW_COUNT = 600;
34const OUTPUT_BYTES_PER_ROW = 8_192;
35
36// Recorded before server-resolved dispatch and raw tool-output wire omission at
37// 91b193653ec. The old path read this entire response once for every routine
38// implicit-auto message send; keep the raw fixture to preserve that baseline.
39const HISTORICAL_FULL_PROJECTION_APPLICATION_BYTES = 10_371_419;
40// Projection response + dispatch command + dispatch receipt, excluding RPC
41// envelope bytes. The omitted projection request only makes the old path larger.
42const HISTORICAL_FORMER_SEND_APPLICATION_BYTES = 10_371_678;
43// Corrected measurement: contract-encoded payloads in Effect RPC Request/Exit
44// envelopes before WebSocket framing and compression. Trace values use the
45// production 32/16 hex-character widths so the byte cost is representative.
46const FORMER_SEND_RPC_JSON_BYTES = 10_372_184;
47// The corrected first server-resolved implementation measured 522 bytes.
48const MAX_CURRENT_SEND_RPC_JSON_BYTES = 1_024;
49
50const RPC_TRACE_CONTEXT = {
51 traceId: "0".repeat(32),
52 spanId: "0".repeat(16),
53 sampled: true,
54} as const;
55
56const encodeGetThreadProjectionInput = Schema.encodeSync(
57 Schema.toCodecJson(OrchestrationV2RpcSchemas.getThreadProjection.input),
58);
59const encodeGetThreadProjectionOutput = Schema.encodeSync(
60 Schema.toCodecJson(OrchestrationV2RpcSchemas.getThreadProjection.output),
61);
62const encodeDispatchCommandInput = Schema.encodeSync(
63 Schema.toCodecJson(OrchestrationV2RpcSchemas.dispatchCommand.input),
64);
65const encodeDispatchCommandOutput = Schema.encodeSync(
66 Schema.toCodecJson(OrchestrationV2RpcSchemas.dispatchCommand.output),
67);
68
69function encodedBytes(value: unknown): number {
70 return new TextEncoder().encode(JSON.stringify(value)).byteLength;
71}
72
73function encodedRpcRequestBytes(input: {
74 readonly requestId: number;
75 readonly tag: string;
76 readonly payload: unknown;
77}): number {
78 const request = {
79 _tag: "Request",
80 id: input.requestId,
81 tag: input.tag,
82 payload: input.payload,
83 ...RPC_TRACE_CONTEXT,
84 headers: [],
85 } satisfies RpcMessage.RequestEncoded;
86 return encodedBytes(request);
87}
88
89function encodedRpcSuccessBytes(input: {
90 readonly requestId: number;
91 readonly value: unknown;
92}): number {
93 const response = {
94 _tag: "Exit",
95 requestId: input.requestId,
96 exit: { _tag: "Success", value: input.value },
97 } satisfies RpcMessage.ResponseExitEncoded;
98 return encodedBytes(response);
99}
100
101function makeLargeProjection(): OrchestrationV2ThreadProjection {
102 const visibleTurnItems: OrchestrationV2ProjectedTurnItem[] = Array.from(
103 { length: ROW_COUNT },
104 (_, index) => {
105 const item: OrchestrationV2TurnItem = {
106 id: TurnItemId.make(`item-${index}`),
107 type: "command_execution",
108 threadId: v2ThreadId,
109 runId: null,
110 nodeId: null,
111 providerThreadId: null,
112 providerTurnId: null,
113 nativeItemRef: null,
114 parentItemId: null,
115 ordinal: index + 1,
116 status: "completed",
117 title: `Command ${index}`,
118 input: `cmd-${index}`,
119 output: "x".repeat(OUTPUT_BYTES_PER_ROW),
120 exitCode: 0,
121 startedAt: v2Now,
122 completedAt: v2Now,
123 updatedAt: v2Now,
124 };
125 return {
126 position: index,
127 visibility: "local",
128 sourceThreadId: v2ThreadId,
129 sourceItemId: item.id,
130 item,
131 };
132 },
133 );
134
135 return {
136 ...v2Projection,
137 turnItems: visibleTurnItems.map((row) => row.item),
138 visibleTurnItems,
139 };
140}
141
142const TARGET = new PrimaryConnectionTarget({
143 environmentId: EnvironmentId.make("environment-perf"),
144 label: "Performance fixture",
145 httpBaseUrl: "https://environment.example.test",
146 wsBaseUrl: "wss://environment.example.test",
147});
148
149const layerTestCrypto = Layer.succeed(
150 Crypto.Crypto,
151 Crypto.make({
152 randomBytes: (size) => new Uint8Array(size),
153 digest: (_algorithm, data) => Effect.succeed(data),
154 }),
155);
156
157const makeSupervisor = Effect.fn("CommandPerformance.makeSupervisor")(function* (input: {
158 readonly projection: OrchestrationV2ThreadProjection;
159 readonly projectionRequests: Array<unknown>;
160 readonly projectionResponseBytes: { value: number };
161 readonly commands: OrchestrationV2Command[];
162}) {
163 const client = {
164 [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (requestInput: unknown) =>
165 Effect.sync(() => {
166 input.projectionRequests.push(requestInput);
167 input.projectionResponseBytes.value += encodedBytes(input.projection);
168 return input.projection;
169 }),
170 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
171 Effect.sync(() => {
172 input.commands.push(command);
173 return { sequence: input.commands.length };
174 }),
175 } as unknown as WsRpcProtocolClient;
176 const session: RpcSession.RpcSession = {
177 client,
178 initialConfig: Effect.succeed({
179 environment: {
180 capabilities: {
181 repositoryIdentity: true,
182 serverResolvedCommandContext: true,
183 },
184 },
185 } as never),
186 subscribeServerConfig: (requestInput) => client.subscribeServerConfig(requestInput),
187 ready: Effect.void,
188 probe: Effect.void,
189 closed: Effect.never,
190 };
191 return EnvironmentSupervisor.EnvironmentSupervisor.of({
192 target: TARGET,
193 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
194 session: yield* SubscriptionRef.make(Option.some(session)),
195 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
196 connect: Effect.void,
197 disconnect: Effect.void,
198 retryNow: Effect.void,
199 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
200});
201
202describe("routine command transport budget", () => {
203 it.effect("dispatches an implicit-auto message without a full projection round trip", () =>
204 Effect.gen(function* () {
205 const projection = makeLargeProjection();
206 const projectionRequests: unknown[] = [];
207 const projectionResponseBytes = { value: 0 };
208 const commands: OrchestrationV2Command[] = [];
209 const supervisor = yield* makeSupervisor({
210 projection,
211 projectionRequests,
212 projectionResponseBytes,
213 commands,
214 });
215
216 const receipt = yield* startThreadTurn({
217 commandId: CommandId.make("perf-send"),
218 threadId: v2ThreadId,
219 message: {
220 messageId: MessageId.make("message-perf"),
221 role: "user",
222 text: "continue",
223 attachments: [],
224 },
225 runtimeMode: "full-access",
226 interactionMode: "default",
227 titleSeed: "continue",
228 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
229
230 const command = commands[0]!;
231 if (command.type !== "message.dispatch") {
232 throw new Error(`Expected message.dispatch, received ${command.type}`);
233 }
234 if (!("sequence" in receipt)) {
235 throw new Error("Expected a dispatch receipt");
236 }
237 const { deliveryIntent: _deliveryIntent, ...formerCommand } = command;
238
239 const historicalApplicationBytes =
240 encodedBytes(projection) + encodedBytes(formerCommand) + encodedBytes(receipt);
241 const projectionRequestRpcJsonBytes = encodedRpcRequestBytes({
242 requestId: 0,
243 tag: ORCHESTRATION_V2_WS_METHODS.getThreadProjection,
244 payload: encodeGetThreadProjectionInput({ threadId: v2ThreadId }),
245 });
246 const projectionResponseRpcJsonBytes = encodedRpcSuccessBytes({
247 requestId: 0,
248 value: encodeGetThreadProjectionOutput(projection),
249 });
250 const formerDispatchRequestRpcJsonBytes = encodedRpcRequestBytes({
251 requestId: 1,
252 tag: ORCHESTRATION_V2_WS_METHODS.dispatchCommand,
253 payload: encodeDispatchCommandInput(formerCommand),
254 });
255 const currentDispatchRequestRpcJsonBytes = encodedRpcRequestBytes({
256 requestId: 1,
257 tag: ORCHESTRATION_V2_WS_METHODS.dispatchCommand,
258 payload: encodeDispatchCommandInput(command),
259 });
260 const receiptRpcJsonBytes = encodedRpcSuccessBytes({
261 requestId: 1,
262 value: encodeDispatchCommandOutput(receipt),
263 });
264 const formerRpcJsonBytes =
265 projectionRequestRpcJsonBytes +
266 projectionResponseRpcJsonBytes +
267 formerDispatchRequestRpcJsonBytes +
268 receiptRpcJsonBytes;
269 const currentRpcJsonBytes = currentDispatchRequestRpcJsonBytes + receiptRpcJsonBytes;
270
271 expect(encodedBytes(projection)).toBe(HISTORICAL_FULL_PROJECTION_APPLICATION_BYTES);
272 expect(historicalApplicationBytes).toBe(HISTORICAL_FORMER_SEND_APPLICATION_BYTES);
273 expect(formerRpcJsonBytes).toBe(FORMER_SEND_RPC_JSON_BYTES);
274 expect(projectionRequests).toEqual([]);
275 expect(projectionResponseBytes.value).toBe(0);
276 expect(command).toMatchObject({
277 type: "message.dispatch",
278 deliveryIntent: "auto",
279 dispatchMode: { type: "start_immediately" },
280 });
281 expect(currentRpcJsonBytes).toBeLessThanOrEqual(MAX_CURRENT_SEND_RPC_JSON_BYTES);
282 expect(currentRpcJsonBytes / formerRpcJsonBytes).toBeLessThanOrEqual(0.000_06);
283 }).pipe(Effect.provide(layerTestCrypto)),
284 );
285});