1import * as NodeModule from "node:module";
2import * as NodeURL from "node:url";
3import * as NodeCrypto from "node:crypto";
4import * as NodeAssert from "node:assert/strict";
5
6const root = process.argv[2];
7const mode = process.argv[3];
8const count = Number(process.argv[4] ?? 4);
9const bytes = Number(process.argv[5] ?? 32768);
10const require = NodeModule.createRequire(root + "/apps/server/package.json");
11const load = (name) => import(NodeURL.pathToFileURL(require.resolve("effect/" + name)));
12const [Effect, Stream, Layer, Deferred, Fiber] = await Promise.all(
13 ["Effect", "Stream", "Layer", "Deferred", "Fiber"].map(load),
14);
15const app = (file) => import(NodeURL.pathToFileURL(root + "/apps/server/src/" + file + ".ts"));
16const [Prefix, Ws, Threads, Events] = await Promise.all([
17 app("rpcInitialItems"),
18 app("ws"),
19 app("orchestration-v2/ThreadManagementService"),
20 app("persistence/OrchestrationEventStore"),
21]);
22const refs = [];
23const fibers = [];
24const checkpoints = [];
25const deliveredCounts = [];
26
27function history() {
28 const value = JSON.parse(JSON.stringify({ text: NodeCrypto.randomBytes(bytes).toString("hex") }));
29 refs.push(new WeakRef(value));
30 return value;
31}
32
33function start(ready) {
34 const index = deliveredCounts.push(0) - 1;
35 const live = Stream.fromEffect(Deferred.succeed(ready, undefined)).pipe(
36 Stream.drain,
37 Stream.concat(Stream.never),
38 );
39 const dependencies = Layer.mergeAll(
40 Layer.mock(Threads.ThreadManagementService)({
41 ensureLegacyTranscript: () => Effect.void,
42 getThreadSnapshot: () =>
43 Effect.sync(() => ({
44 snapshotSequence: 1,
45 projection: {
46 messages: [history()],
47 contextHandoffs: [],
48 turnItems: [],
49 visibleTurnItems: [],
50 },
51 })),
52 streamStoredEventsFrom: () => live,
53 }),
54 Layer.mock(Events.OrchestrationEventStore)({
55 latestAgentSequence: () => Effect.succeed(1),
56 getAgentReplayStats: () =>
57 Effect.succeed({
58 eventCount: 1,
59 rawPayloadBytes: bytes * 2,
60 hasCreateEvent: false,
61 }),
62 readAgentEvents: () =>
63 Stream.fromEffect(
64 Effect.sync(() => ({
65 sequence: 1,
66 event: { type: "message.updated", payload: history() },
67 })),
68 ),
69 }),
70 );
71 const stream =
72 mode === "prefix"
73 ? Effect.succeed(Stream.concat(Prefix.rpcInitialItems([history()]), live))
74 : Ws.subscribeOrchestrationV2Thread({
75 threadId: "synthetic-thread",
76 requestCompletionMarker: true,
77 ...(mode === "replay" ? { afterSequence: 0 } : {}),
78 }).pipe(Effect.provide(dependencies));
79 let delivered = 0;
80 return stream.pipe(
81 Effect.flatMap((stream) =>
82 stream.pipe(
83 Stream.runForEach((item) =>
84 Effect.sync(() => {
85 if (mode === "prefix") {
86 NodeAssert.equal(item.text.length, bytes * 2);
87 } else if (delivered === 0) {
88 NodeAssert.equal(item.kind, mode === "replay" ? "event" : "snapshot");
89 const value = mode === "replay" ? item.event.payload : item.projection.messages[0];
90 NodeAssert.equal(value.text.length, bytes * 2);
91 } else {
92 NodeAssert.equal(item.kind, "synchronized");
93 }
94 delivered++;
95 deliveredCounts[index] = delivered;
96 }),
97 ),
98 Effect.forkDetach,
99 ),
100 ),
101 );
102}
103
104const result = await Effect.runPromise(
105 Effect.gen(function* () {
106 for (let i = 0; i < count; i++) {
107 const ready = yield* Deferred.make();
108 fibers.push(yield* start(ready));
109 yield* Deferred.await(ready);
110 NodeAssert.equal(deliveredCounts[i], mode === "prefix" ? 1 : 2);
111 if (i === Math.floor(count / 2) - 1 || i === count - 1) {
112 yield* Effect.promise(async () => {
113 for (let j = 0; j < 3; j++) {
114 await new Promise(setImmediate);
115 global.gc();
116 }
117 checkpoints.push({
118 subscriptions: fibers.length,
119 retained: refs.filter((ref) => ref.deref()).length,
120 payloadMiB: ((i + 1) * bytes * 2) / 1048576,
121 heapMiB: Math.round(process.memoryUsage().heapUsed / 1048576),
122 });
123 });
124 }
125 }
126 for (const fiber of fibers) yield* Fiber.interrupt(fiber);
127 return { checkpoints };
128 }),
129);
130console.log(JSON.stringify(result));