1import { unwrapRpcHandlers } from "@/Local/RpcSerialization.ts";
2import type { RpcProxyApi } from "@/Local/RpcServer.ts";
3import {
4 encodeSessionEnvironment,
5 SESSION_ENV_PARAM,
6} from "@/Local/RpcServerEnvironment.ts";
7import {
8 layerServer,
9 RpcSpawner,
10 LOGS_PATH,
11 parseSidecarLogLine,
12 type RpcSpawnPayload,
13} from "@/Local/RpcSpawner.ts";
14import { PlatformServices } from "@/Util/PlatformServices.ts";
15import { describe, expect, it } from "alchemy-test";
16import { newWebSocketRpcSession, type RpcStub } from "capnweb";
17import * as Duration from "effect/Duration";
18import * as Effect from "effect/Effect";
19import * as Fiber from "effect/Fiber";
20import * as Layer from "effect/Layer";
21import * as Schedule from "effect/Schedule";
22import * as Stream from "effect/Stream";
23import * as FetchHttpClient from "effect/http/FetchHttpClient";
24import * as HttpBody from "effect/http/HttpBody";
25import * as HttpClient from "effect/http/HttpClient";
26import * as HttpClientRequest from "effect/http/HttpClientRequest";
27import {
28 assertPidExited,
29 canOpenWebSocket,
30 isAlive,
31 openWebSocket,
32 pgidOf,
33 pidListeningOn,
34} from "./fixtures/process-effect.ts";
35
36const FIXTURE_TS_URL = new URL(
37 "./fixtures/rpc-server-entry.ts",
38 import.meta.url,
39).toString();
40const FIXTURE_B_TS_URL = new URL(
41 "./fixtures/rpc-server-entry-b.ts",
42 import.meta.url,
43).toString();
44const CRASH_FIXTURE_TS_URL = new URL(
45 "./fixtures/rpc-server-crash.ts",
46 import.meta.url,
47).toString();
48const LOGS_FIXTURE_TS_URL = new URL(
49 "./fixtures/rpc-server-logs.ts",
50 import.meta.url,
51).toString();
52
53const samplePayload = (serverEntryUrl: string): RpcSpawnPayload => ({
54 serverEntryUrl,
55});
56
57// The spawner inherits the runtime that vitest itself is running under
58// (it shells out to `bun` or `node` based on `typeof Bun`). These tests
59// only verify behavior for the active runtime; run vitest under both to
60// get full coverage.
61describe(
62 `Local.RpcSpawner (runtime=${typeof globalThis.Bun !== "undefined" ? "bun" : "node"})`,
63 { tags: ["local"] },
64 () => {
65 /**
66 * The Spawner layer (and any child processes it spawns) is torn down
67 * when the surrounding test scope closes, so we provide it at the test
68 * boundary rather than wrapping a sub-effect — that would tear the
69 * server down the moment the sub-effect returned.
70 */
71 const services = Layer.provideMerge(
72 layerServer({ profile: undefined, envFile: undefined }),
73 Layer.merge(PlatformServices, FetchHttpClient.layer),
74 );
75
76 it.live(
77 "POST returns a ws url whose RPC end-to-end call hits the fixture",
78 () =>
79 Effect.gen(function* () {
80 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
81 const wsUrl = yield* post(url, samplePayload(FIXTURE_TS_URL));
82 expect(wsUrl).toMatch(/^ws:\/\//);
83 const result = yield* echoWebSocket(wsUrl, "hello");
84 expect(result).toBe("echo:hello");
85 }).pipe(Effect.provide(services)),
86 { timeout: 60_000 },
87 );
88
89 it.live(
90 "rejects malformed spawn payloads and non-file URLs",
91 () =>
92 Effect.gen(function* () {
93 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
94 for (const payload of [
95 {},
96 { serverEntryUrl: 42 },
97 { serverEntryUrl: "not a url" },
98 { serverEntryUrl: "https://example.com/server.ts" },
99 ]) {
100 const response = yield* postRaw(url, payload);
101 expect(response.status).toBe(400);
102 }
103 }).pipe(Effect.provide(services)),
104 { timeout: 60_000 },
105 );
106
107 it(
108 "client parser ignores heartbeat frames",
109 () => {
110 expect(parseSidecarLogLine('{"channel":"heartbeat"}')).toBeUndefined();
111 },
112 { tags: ["unit"] },
113 );
114
115 it.live(
116 "emits a heartbeat immediately and periodically",
117 () =>
118 Effect.gen(function* () {
119 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
120 const client = yield* HttpClient.HttpClient;
121 const response = yield* client.get(
122 new URL(LOGS_PATH, url).toString(),
123 );
124 const frames = yield* response.stream.pipe(
125 Stream.decodeText,
126 Stream.splitLines,
127 Stream.map((line) => JSON.parse(line) as { channel: string }),
128 Stream.filter(({ channel }) => channel === "heartbeat"),
129 Stream.take(2),
130 Stream.runCollect,
131 Effect.timeout(Duration.seconds(7)),
132 );
133 expect(Array.from(frames)).toEqual([
134 { channel: "heartbeat" },
135 { channel: "heartbeat" },
136 ]);
137 }).pipe(Effect.provide(services)),
138 { timeout: 10_000 },
139 );
140
141 it.live(
142 "caches the child by entry url: a second POST returns the same url",
143 () =>
144 Effect.gen(function* () {
145 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
146 const payload = samplePayload(FIXTURE_TS_URL);
147 const first = yield* post(url, payload);
148 const second = yield* post(url, payload);
149 expect(second).toBe(first);
150 const pid = yield* pidListeningOn(first);
151 if (pid !== undefined) {
152 expect(yield* isAlive(pid)).toBe(true);
153 }
154 }).pipe(Effect.provide(services)),
155 { timeout: 60_000 },
156 );
157
158 it.live(
159 "distinct entry urls spawn distinct children with distinct urls",
160 () =>
161 Effect.gen(function* () {
162 // Children are keyed by serverEntryUrl ONLY — stacks share one child
163 // (each RPC session carries its own stack environment), so the second
164 // fixture must be a genuinely different entry module.
165 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
166 const a = yield* post(url, samplePayload(FIXTURE_TS_URL));
167 const b = yield* post(url, samplePayload(FIXTURE_B_TS_URL));
168 expect(a).not.toBe(b);
169 }).pipe(Effect.provide(services)),
170 { timeout: 60_000 },
171 );
172
173 it.live(
174 "sidecars lead their own process group (a terminal Ctrl+C cannot reach them)",
175 () =>
176 Effect.gen(function* () {
177 // POSIX-only: Windows has no process groups; the spawner leaves
178 // sidecars attached there (detached console children have their own
179 // quirks) and Ctrl+C delivery works differently anyway.
180 if (process.platform === "win32") return;
181 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
182 const wsUrl = yield* post(url, samplePayload(FIXTURE_TS_URL));
183 const pid = yield* pidListeningOn(wsUrl);
184 if (pid === undefined || Number.isNaN(pid)) return;
185 // Group leader (pgid === pid) proves the sidecar was spawned
186 // detached: the tty's foreground-group signals stop at this process.
187 expect(yield* pgidOf(pid)).toBe(pid);
188 }).pipe(Effect.provide(services)),
189 { timeout: 60_000 },
190 );
191
192 it.live(
193 "closing the spawner's scope kills all spawned children",
194 () =>
195 Effect.gen(function* () {
196 // Boot the spawner in an inner scope so we can close it while
197 // the outer test scope is still alive, then assert against the
198 // pid we recorded.
199 const pid = yield* Effect.gen(function* () {
200 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
201 const wsUrl = yield* post(url, samplePayload(FIXTURE_TS_URL));
202 return yield* pidListeningOn(wsUrl);
203 }).pipe(Effect.provide(services), Effect.scoped);
204
205 if (pid === undefined) return;
206 yield* assertPidExited(pid);
207 }),
208 { timeout: 60_000 },
209 );
210
211 it.live(
212 "url returned for a crash-on-boot fixture is not a usable RPC endpoint",
213 () =>
214 Effect.gen(function* () {
215 // The crash fixture prints the address marker then exits. The
216 // spawner's health check is best-effort: depending on race
217 // timing the POST may return a bogus url, or surface a 500
218 // once the retry budget drains. The invariant we *can*
219 // assert is that callers cannot open a parent websocket to
220 // the returned url.
221 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
222 yield* Effect.gen(function* () {
223 const r = yield* postRaw(url, samplePayload(CRASH_FIXTURE_TS_URL));
224 if (r.status !== 200) {
225 return { unusable: true } as const;
226 }
227 const usable = yield* canOpenWebSocket(new URL("/parent", r.body));
228 return { unusable: !usable } as const;
229 }).pipe(
230 Effect.flatMap((r) =>
231 r.unusable
232 ? Effect.void
233 : Effect.fail(new Error("endpoint was still usable")),
234 ),
235 // Mirrors the original `for (let i = 0; i < 4 && !failed; i++)`
236 // loop: up to 4 retries spaced 250ms apart.
237 Effect.retry({
238 schedule: Schedule.spaced(Duration.millis(250)),
239 times: 4,
240 }),
241 );
242 }).pipe(Effect.provide(services)),
243 { timeout: 60_000 },
244 );
245
246 it.live(
247 "a /logs subscriber receives sidecar stdout and stderr as ndjson",
248 () =>
249 Effect.gen(function* () {
250 const url = yield* RpcSpawner.useSync((spawner) => spawner.url);
251 const client = yield* HttpClient.HttpClient;
252
253 // Subscribe BEFORE spawning (headers received = subscription
254 // registered server-side) — with a subscriber connected, the
255 // spawner must route sidecar output here instead of its console.
256 const response = yield* client.get(
257 new URL(LOGS_PATH, url).toString(),
258 );
259 const collector = yield* response.stream.pipe(
260 Stream.decodeText,
261 Stream.splitLines,
262 Stream.map(
263 (line) => JSON.parse(line) as { channel: string; line?: string },
264 ),
265 // drop heartbeats (no `line`) and any non-fixture noise
266 Stream.filter(
267 (entry): entry is { channel: string; line: string } =>
268 entry.line?.startsWith("fixture-") ?? false,
269 ),
270 Stream.take(6),
271 Stream.runCollect,
272 Effect.forkScoped,
273 );
274
275 const wsUrl = yield* post(url, samplePayload(LOGS_FIXTURE_TS_URL));
276 expect(wsUrl).toMatch(/^ws:\/\//);
277
278 const received = Array.from(
279 yield* Fiber.join(collector).pipe(
280 Effect.timeout(Duration.seconds(20)),
281 ),
282 );
283 const channels = new Set(received.map((entry) => entry.channel));
284 expect(
285 received.every((entry) => entry.line.startsWith("fixture-")),
286 ).toBe(true);
287 expect(channels.has("stdout")).toBe(true);
288 expect(channels.has("stderr")).toBe(true);
289 }).pipe(Effect.provide(services)),
290 { timeout: 60_000 },
291 );
292 },
293);
294
295interface PostResult {
296 readonly status: number;
297 readonly body: string;
298}
299
300const postRaw = (
301 url: string,
302 body: unknown,
303): Effect.Effect<PostResult, never, HttpClient.HttpClient> =>
304 Effect.gen(function* () {
305 const client = yield* HttpClient.HttpClient;
306 const req = HttpClientRequest.post(url).pipe(
307 HttpClientRequest.setBody(
308 HttpBody.text(JSON.stringify(body), "application/json"),
309 ),
310 );
311 const res = yield* client.execute(req);
312 const text = yield* res.text;
313 return { status: res.status, body: text };
314 }).pipe(Effect.orDie);
315
316const post = (
317 url: string,
318 body: unknown,
319): Effect.Effect<string, Error, HttpClient.HttpClient> =>
320 postRaw(url, body).pipe(
321 Effect.flatMap((r) =>
322 r.status === 200
323 ? Effect.succeed(r.body)
324 : Effect.fail(new Error(`spawn POST failed: ${r.status} ${r.body}`)),
325 ),
326 );
327
328const echoWebSocket = (
329 rpcUrl: string,
330 msg: string,
331): Effect.Effect<string, Error> =>
332 Effect.gen(function* () {
333 yield* openWebSocket(new URL("/parent", rpcUrl));
334 // Sessions carry their stack environment (real clients — the
335 // RpcProviderProxy — always send one; a session without it is an error).
336 const sessionUrl = new URL(rpcUrl);
337 sessionUrl.searchParams.set(
338 SESSION_ENV_PARAM,
339 encodeSessionEnvironment({
340 alchemyContext: {
341 dotAlchemy: "/tmp/.alchemy",
342 dev: true,
343 adopt: false,
344 },
345 stack: { name: "test", stage: "dev" },
346 }),
347 );
348 return yield* Effect.promise(async () => {
349 // Cast through `unknown`: comparing capnweb's deeply-recursive Stub
350 // type against RpcStub<RpcProxyApi> exceeds the compiler's
351 // instantiation depth (TS2589/TS2321).
352 const stub = newWebSocketRpcSession(
353 sessionUrl.toString(),
354 ) as unknown as RpcStub<RpcProxyApi>;
355 const provider = await stub.getProvider("Test.Echo", FIXTURE_TS_URL);
356 const handlers = unwrapRpcHandlers(provider as any) as {
357 echo: (m: string) => Effect.Effect<string>;
358 };
359 return await Effect.runPromise(handlers.echo(msg));
360 });
361 }).pipe(Effect.scoped);