1import {
2 EnvironmentId,
3 EventId,
4 MessageId,
5 type OrchestrationV2DomainEvent,
6 type OrchestrationV2ThreadProjection,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Option from "effect/Option";
10import { HttpClient, HttpClientResponse } from "effect/http";
11import { describe, test } from "vite-plus/test";
12
13import { issueRemoteWebSocketTicket } from "./authorization/remote.ts";
14import { PrimaryConnectionTarget } from "./connection/model.ts";
15import { fetchRemoteEnvironmentDescriptor } from "./environment/descriptor.ts";
16import type { RemoteEnvironmentRequestError } from "./rpc/http.ts";
17import { fetchEnvironmentThreadSnapshot } from "./state/threadSnapshotHttp.ts";
18import { applyOrchestrationV2ProjectionEvent } from "./state/orchestrationV2Projection.ts";
19import { v2Projection, v2Now } from "./state/orchestrationV2TestFixtures.ts";
20
21const timestamp = "2026-09-01T00:00:00.000Z";
22const runOptions = { warmupTime: 1_000, time: 1_500 };
23const thread: OrchestrationV2ThreadProjection = {
24 ...v2Projection,
25 messages: Array.from({ length: 100 }, (_, index) => ({
26 id: MessageId.make(`message-${index}`),
27 threadId: v2Projection.thread.id,
28 runId: null,
29 nodeId: null,
30 role: "assistant",
31 text: "Message text. ".repeat(40),
32 attachments: [],
33 streaming: false,
34 createdBy: "agent",
35 creationSource: "provider",
36 createdAt: v2Now,
37 updatedAt: v2Now,
38 })),
39};
40const target = new PrimaryConnectionTarget({
41 environmentId: EnvironmentId.make("remote-1"),
42 label: "Remote",
43 httpBaseUrl: "https://remote.example.test",
44 wsBaseUrl: "wss://remote.example.test/ws",
45});
46const responses = {
47 "/.well-known/t3/environment": {
48 environmentId: target.environmentId,
49 label: target.label,
50 platform: { os: "linux", arch: "x64" },
51 serverVersion: "0.0.0-test",
52 capabilities: { repositoryIdentity: true },
53 },
54 "/api/auth/websocket-ticket": { ticket: "test-ticket", expiresAt: timestamp },
55 "/api/orchestration/threads/thread-v2": { snapshotSequence: 1, projection: thread },
56};
57const httpClient = HttpClient.make((request) =>
58 Effect.sync(() => {
59 const path = new URL(request.url).pathname as keyof typeof responses;
60 return HttpClientResponse.fromWeb(request, Response.json(responses[path]));
61 }),
62);
63const requests: Record<
64 string,
65 Effect.Effect<unknown, RemoteEnvironmentRequestError, HttpClient.HttpClient>
66> = {
67 "read remote connection descriptor": fetchRemoteEnvironmentDescriptor({
68 httpBaseUrl: target.httpBaseUrl,
69 }),
70 "issue remote WebSocket ticket": issueRemoteWebSocketTicket({
71 httpBaseUrl: target.httpBaseUrl,
72 bearerToken: "test-token",
73 }),
74 "load remote snapshot with 100 messages": fetchEnvironmentThreadSnapshot({
75 prepared: {
76 environmentId: target.environmentId,
77 label: target.label,
78 httpBaseUrl: target.httpBaseUrl,
79 socketUrl: target.wsBaseUrl,
80 httpAuthorization: null,
81 target,
82 },
83 threadId: thread.thread.id,
84 signer: Option.none(),
85 }),
86};
87
88describe("remote HTTP processing with an in-memory transport", () => {
89 for (const [name, request] of Object.entries(requests)) {
90 test(name, async ({ bench }) => {
91 await bench(name, async () => {
92 await Effect.runPromise(
93 request.pipe(Effect.provideService(HttpClient.HttpClient, httpClient)),
94 );
95 }).run(runOptions);
96 });
97 }
98});
99
100const delta: Extract<OrchestrationV2DomainEvent, { type: "message.updated" }> = {
101 id: EventId.make("delta"),
102 type: "message.updated",
103 threadId: thread.thread.id,
104 occurredAt: v2Now,
105 payload: { ...thread.messages[99]!, text: " next", streaming: true },
106};
107
108describe("remote message replay", () => {
109 for (const count of [100, 1_000]) {
110 const loaded = {
111 ...thread,
112 messages: Array.from({ length: count }, (_, index) => ({
113 ...thread.messages[0]!,
114 id: MessageId.make(`message-${index}`),
115 })),
116 };
117 const event = {
118 ...delta,
119 payload: { ...delta.payload, id: loaded.messages.at(-1)!.id },
120 };
121 const name = `apply 200 message updates to ${count} loaded messages`;
122 test(name, async ({ bench }) => {
123 await bench(name, () => {
124 let current: OrchestrationV2ThreadProjection = loaded;
125 for (let index = 0; index < 200; index += 1) {
126 current =
127 applyOrchestrationV2ProjectionEvent(current, {
128 ...event,
129 payload: { ...event.payload, text: ` next ${index}` },
130 }) ?? current;
131 }
132 }).run(runOptions);
133 });
134 }
135});