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 SecureRandom from "effect/crypto/SecureRandom";
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
36const HISTORICAL_FULL_PROJECTION_APPLICATION_BYTES = 10_371_419;
37const HISTORICAL_FORMER_SEND_APPLICATION_BYTES = 10_371_678;
38const FORMER_SEND_RPC_JSON_BYTES = 10_372_184;
39const MAX_CURRENT_SEND_RPC_JSON_BYTES = 1_024;
40
41const RPC_TRACE_CONTEXT = {
42 traceId: "0".repeat(32),
43 spanId: "0".repeat(16),
44 sampled: true,
45} as const;
46
47const encodeGetThreadProjectionInput = Schema.encodeSync(
48 Schema.toCodecJson(OrchestrationV2RpcSchemas.getThreadProjection.input),
49);
50const encodeGetThreadProjectionOutput = Schema.encodeSync(
51 Schema.toCodecJson(OrchestrationV2RpcSchemas.getThreadProjection.output),
52);
53const encodeDispatchCommandInput = Schema.encodeSync(
54 Schema.toCodecJson(OrchestrationV2RpcSchemas.dispatchCommand.input),
55);
56const encodeDispatchCommandOutput = Schema.encodeSync(
57 Schema.toCodecJson(OrchestrationV2RpcSchemas.dispatchCommand.output),
58);
59
60function encodedBytes(value: unknown): number {
61 return new TextEncoder().encode(JSON.stringify(value)).byteLength;
62}
63
64function encodedRpcRequestBytes(input: {
65 readonly requestId: number;
66 readonly tag: string;
67 readonly payload: unknown;
68}): number {
69 const request = {
70 _tag: "Request",
71 id: input.requestId,
72 tag: input.tag,
73 payload: input.payload,
74 ...RPC_TRACE_CONTEXT,
75 headers: [],
76 } satisfies RpcMessage.RequestEncoded;
77 return encodedBytes(request);
78}
79
80function encodedRpcSuccessBytes(input: {
81 readonly requestId: number;
82 readonly value: unknown;
83}): number {
84 const response = {
85 _tag: "Exit",
86 requestId: input.requestId,
87 exit: { _tag: "Success", value: input.value },
88 } satisfies RpcMessage.ResponseExitEncoded;
89 return encodedBytes(response);
90}
91
92function makeLargeProjection(): OrchestrationV2ThreadProjection {
93 const visibleTurnItems: OrchestrationV2ProjectedTurnItem[] = Array.from(
94 { length: ROW_COUNT },
95 (_, index) => {
96 const item: OrchestrationV2TurnItem = {
97 id: TurnItemId.make(`item-${index}`),
98 type: "command_execution",
99 threadId: v2ThreadId,
100 runId: null,
101 nodeId: null,
102 providerThreadId: null,
103 providerTurnId: null,
104 nativeItemRef: null,
105 parentItemId: null,
106 ordinal: index + 1,
107 status: "completed",
108 title: `Command ${index}`,
109 input: `cmd-${index}`,
110 output: "x".repeat(OUTPUT_BYTES_PER_ROW),
111 exitCode: 0,
112 startedAt: v2Now,
113 completedAt: v2Now,
114 updatedAt: v2Now,
115 };
116 return {
117 position: index,
118 visibility: "local",
119 sourceThreadId: v2ThreadId,
120 sourceItemId: item.id,
121 item,
122 };
123 },
124 );
125
126 return {
127 ...v2Projection,
128 turnItems: visibleTurnItems.map((row) => row.item),
129 visibleTurnItems,
130 };
131}
132
133const TARGET = new PrimaryConnectionTarget({
134 environmentId: EnvironmentId.make("environment-perf"),
135 label: "Performance fixture",
136 httpBaseUrl: "https://environment.example.test",
137 wsBaseUrl: "wss://environment.example.test",
138});
139
140const layerTestRandom = Layer.succeed(
141 SecureRandom.SecureRandom,
142 SecureRandom.make((size) => new Uint8Array(size)),
143);
144
145const makeSupervisor = Effect.fn("CommandPerformance.makeSupervisor")(function* (input: {
146 readonly projection: OrchestrationV2ThreadProjection;
147 readonly projectionRequests: Array<unknown>;
148 readonly projectionResponseBytes: { value: number };
149 readonly commands: OrchestrationV2Command[];
150}) {
151 const client = {
152 [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (requestInput: unknown) =>
153 Effect.sync(() => {
154 input.projectionRequests.push(requestInput);
155 input.projectionResponseBytes.value += encodedBytes(input.projection);
156 return input.projection;
157 }),
158 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
159 Effect.sync(() => {
160 input.commands.push(command);
161 return { sequence: input.commands.length };
162 }),
163 } as unknown as WsRpcProtocolClient;
164 const session: RpcSession.RpcSession = {
165 client,
166 initialConfig: Effect.succeed({
167 environment: {
168 capabilities: {
169 repositoryIdentity: true,
170 serverResolvedCommandContext: true,
171 },
172 },
173 } as never),
174 subscribeServerConfig: (requestInput = {}) => client.subscribeServerConfig(requestInput),
175 ready: Effect.void,
176 probe: Effect.void,
177 inboundFrames: Effect.succeed(0),
178 closed: Effect.never,
179 };
180 return EnvironmentSupervisor.EnvironmentSupervisor.of({
181 target: TARGET,
182 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
183 session: yield* SubscriptionRef.make(Option.some(session)),
184 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
185 connect: Effect.void,
186 disconnect: Effect.void,
187 retryNow: Effect.void,
188 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
189});
190
191describe("routine command transport budget", () => {
192 it.effect("dispatches an implicit-auto message without a full projection round trip", () =>
193 Effect.gen(function* () {
194 const projection = makeLargeProjection();
195 const projectionRequests: unknown[] = [];
196 const projectionResponseBytes = { value: 0 };
197 const commands: OrchestrationV2Command[] = [];
198 const supervisor = yield* makeSupervisor({
199 projection,
200 projectionRequests,
201 projectionResponseBytes,
202 commands,
203 });
204
205 const receipt = yield* startThreadTurn({
206 commandId: CommandId.make("perf-send"),
207 threadId: v2ThreadId,
208 message: {
209 messageId: MessageId.make("message-perf"),
210 role: "user",
211 text: "continue",
212 attachments: [],
213 },
214 runtimeMode: "full-access",
215 interactionMode: "default",
216 titleSeed: "continue",
217 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
218
219 const command = commands[0]!;
220 if (command.type !== "message.dispatch") {
221 throw new Error(`Expected message.dispatch, received ${command.type}`);
222 }
223 if ("projection" in receipt) {
224 throw new Error("Expected a dispatch receipt");
225 }
226 const { deliveryIntent: _deliveryIntent, ...formerCommand } = command;
227
228 const historicalApplicationBytes =
229 encodedBytes(projection) + encodedBytes(formerCommand) + encodedBytes(receipt);
230 const projectionRequestRpcJsonBytes = encodedRpcRequestBytes({
231 requestId: 0,
232 tag: ORCHESTRATION_V2_WS_METHODS.getThreadProjection,
233 payload: encodeGetThreadProjectionInput({ threadId: v2ThreadId }),
234 });
235 const projectionResponseRpcJsonBytes = encodedRpcSuccessBytes({
236 requestId: 0,
237 value: encodeGetThreadProjectionOutput(projection),
238 });
239 const formerDispatchRequestRpcJsonBytes = encodedRpcRequestBytes({
240 requestId: 1,
241 tag: ORCHESTRATION_V2_WS_METHODS.dispatchCommand,
242 payload: encodeDispatchCommandInput(formerCommand),
243 });
244 const currentDispatchRequestRpcJsonBytes = encodedRpcRequestBytes({
245 requestId: 1,
246 tag: ORCHESTRATION_V2_WS_METHODS.dispatchCommand,
247 payload: encodeDispatchCommandInput(command),
248 });
249 const receiptRpcJsonBytes = encodedRpcSuccessBytes({
250 requestId: 1,
251 value: encodeDispatchCommandOutput(receipt),
252 });
253 const formerRpcJsonBytes =
254 projectionRequestRpcJsonBytes +
255 projectionResponseRpcJsonBytes +
256 formerDispatchRequestRpcJsonBytes +
257 receiptRpcJsonBytes;
258 const currentRpcJsonBytes = currentDispatchRequestRpcJsonBytes + receiptRpcJsonBytes;
259
260 expect(encodedBytes(projection)).toBe(HISTORICAL_FULL_PROJECTION_APPLICATION_BYTES);
261 expect(historicalApplicationBytes).toBe(HISTORICAL_FORMER_SEND_APPLICATION_BYTES);
262 expect(formerRpcJsonBytes).toBe(FORMER_SEND_RPC_JSON_BYTES);
263 expect(projectionRequests).toEqual([]);
264 expect(projectionResponseBytes.value).toBe(0);
265 expect(command).toMatchObject({
266 type: "message.dispatch",
267 deliveryIntent: "auto",
268 dispatchMode: { type: "start_immediately" },
269 });
270 expect(currentRpcJsonBytes).toBeLessThanOrEqual(MAX_CURRENT_SEND_RPC_JSON_BYTES);
271 expect(currentRpcJsonBytes / formerRpcJsonBytes).toBeLessThanOrEqual(0.000_06);
272 }).pipe(Effect.provide(layerTestRandom)),
273 );
274});