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 { fetchEnvironmentBoundedThreadSnapshot } from "./state/boundedThreadSnapshotHttp.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/bounded": {
56 snapshotSequence: 1,
57 projection: thread,
58 historyCursor: null,
59 hasMoreHistory: false,
60 latestLocalTurnOrdinal: null,
61 },
62};
63const httpClient = HttpClient.make((request) =>
64 Effect.sync(() => {
65 const path = new URL(request.url).pathname as keyof typeof responses;
66 return HttpClientResponse.fromWeb(request, Response.json(responses[path]));
67 }),
68);
69const requests: Record<
70 string,
71 Effect.Effect<unknown, RemoteEnvironmentRequestError, HttpClient.HttpClient>
72> = {
73 "read remote connection descriptor": fetchRemoteEnvironmentDescriptor({
74 httpBaseUrl: target.httpBaseUrl,
75 }),
76 "issue remote WebSocket ticket": issueRemoteWebSocketTicket({
77 httpBaseUrl: target.httpBaseUrl,
78 bearerToken: "test-token",
79 }),
80 "load remote snapshot with 100 messages": fetchEnvironmentBoundedThreadSnapshot({
81 prepared: {
82 environmentId: target.environmentId,
83 label: target.label,
84 httpBaseUrl: target.httpBaseUrl,
85 socketUrl: target.wsBaseUrl,
86 httpAuthorization: null,
87 target,
88 },
89 threadId: thread.thread.id,
90 signer: Option.none(),
91 }),
92};
93
94describe("remote HTTP processing with an in-memory transport", () => {
95 for (const [name, request] of Object.entries(requests)) {
96 test(name, async ({ bench }) => {
97 await bench(name, async () => {
98 await Effect.runPromise(
99 request.pipe(Effect.provideService(HttpClient.HttpClient, httpClient)),
100 );
101 }).run(runOptions);
102 });
103 }
104});
105
106const delta: Extract<OrchestrationV2DomainEvent, { type: "message.updated" }> = {
107 id: EventId.make("delta"),
108 type: "message.updated",
109 threadId: thread.thread.id,
110 occurredAt: v2Now,
111 payload: { ...thread.messages[99]!, text: " next", streaming: true },
112};
113
114describe("remote message replay", () => {
115 for (const count of [100, 1_000]) {
116 const loaded = {
117 ...thread,
118 messages: Array.from({ length: count }, (_, index) => ({
119 ...thread.messages[0]!,
120 id: MessageId.make(`message-${index}`),
121 })),
122 };
123 const event = {
124 ...delta,
125 payload: { ...delta.payload, id: loaded.messages.at(-1)!.id },
126 };
127 const name = `apply 200 message updates to ${count} loaded messages`;
128 test(name, async ({ bench }) => {
129 await bench(name, () => {
130 let current: OrchestrationV2ThreadProjection = loaded;
131 for (let index = 0; index < 200; index += 1) {
132 current =
133 applyOrchestrationV2ProjectionEvent(current, {
134 ...event,
135 payload: { ...event.payload, text: ` next ${index}` },
136 }) ?? current;
137 }
138 }).run(runOptions);
139 });
140 }
141});