1import * as NodeEvents from "node:events";
2import * as NodeNet from "node:net";
3
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { assert, it } from "@effect/vitest";
6import * as HostProcess from "@t3tools/shared/HostProcess";
7import * as Cause from "effect/Cause";
8import * as Effect from "effect/Effect";
9import * as Exit from "effect/Exit";
10import * as Fiber from "effect/Fiber";
11import * as Layer from "effect/Layer";
12import * as Logger from "effect/Logger";
13import * as Scheduler from "effect/Scheduler";
14import { expect, vi } from "vite-plus/test";
15
16import * as NodePtyAdapter from "./NodePtyAdapter.ts";
17import * as PtyAdapter from "@t3tools/shared/PtyAdapter";
18
19function makeNativeProcess(pid = 42) {
20 const events = new NodeEvents.EventEmitter();
21 return {
22 pid,
23 _socket: new NodeNet.Socket(),
24 _agent: { kill: vi.fn() },
25 write: vi.fn(),
26 resize: vi.fn(),
27 kill: vi.fn(),
28 onData: vi.fn((callback: (data: string) => void) => {
29 events.on("data", callback);
30 return {
31 dispose: () => {
32 events.off("data", callback);
33 },
34 };
35 }),
36 onExit: vi.fn((callback: (event: { exitCode: number; signal?: number }) => void) => {
37 events.on("exit", callback);
38 return {
39 dispose: () => {
40 events.off("exit", callback);
41 },
42 };
43 }),
44 events,
45 };
46}
47
48const spawn = vi.fn(() => makeNativeProcess());
49
50function preparePendingProcess() {
51 const nativeProcess = makeNativeProcess(0);
52 const subscribed = Promise.withResolvers<void>();
53 nativeProcess._socket.on("newListener", (event) => {
54 if (event === "ready_datapipe") queueMicrotask(() => subscribed.resolve());
55 });
56 spawn.mockReturnValueOnce(nativeProcess);
57 return { nativeProcess, subscribed: Effect.promise(() => subscribed.promise) };
58}
59
60const spawnInput = { shell: "powershell.exe", cwd: ".", cols: 80, rows: 24, env: {} };
61
62const fakeNodePty = { spawn } as unknown as typeof import("node-pty");
63
64const layerTestFor = (platform: NodeJS.Platform = "win32") =>
65 NodePtyAdapter.layer.pipe(
66 Layer.provide(
67 Layer.mergeAll(
68 NodeServices.layer,
69 Layer.succeed(HostProcess.Platform, platform),
70 Layer.succeed(HostProcess.Architecture, "x64"),
71 Layer.succeed(NodePtyAdapter.NodePtyModuleLoaderRef, () => Promise.resolve(fakeNodePty)),
72 ),
73 ),
74 );
75
76const layerTest = layerTestFor();
77
78it.effect("waits for the Windows PID without requiring output", () =>
79 Effect.gen(function* () {
80 const { nativeProcess, subscribed } = preparePendingProcess();
81 const adapter = yield* PtyAdapter.PtyAdapter;
82 let completed = false;
83 const fiber = yield* adapter.spawn(spawnInput).pipe(
84 Effect.tap(() =>
85 Effect.sync(() => {
86 completed = true;
87 }),
88 ),
89 Effect.forkChild,
90 );
91 yield* subscribed;
92 assert.isFalse(completed);
93 nativeProcess.pid = 12345;
94 nativeProcess._socket.emit("ready_datapipe");
95 const process = yield* Fiber.join(fiber);
96 assert.equal(process.pid, 12345);
97 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
98 assert.equal(nativeProcess.events.listenerCount("exit"), 1);
99
100 const output: string[] = [];
101 const exits: PtyAdapter.PtyExitEvent[] = [];
102 const stopData = process.onData((data) => output.push(data));
103 const stopExit = process.onExit((event) => exits.push(event));
104 nativeProcess.events.emit("data", "first output");
105 nativeProcess.events.emit("exit", { exitCode: 0 });
106 assert.deepEqual(output, ["first output"]);
107 assert.deepEqual(exits, [{ exitCode: 0, signal: null }]);
108 stopData();
109 stopExit();
110 }).pipe(Effect.provide(layerTest)),
111);
112
113it.effect.each(["exit", "close", "error", "invalid-pid"] as const)(
114 "fails Windows startup on %s and cleans up",
115 (failure) =>
116 Effect.gen(function* () {
117 const { nativeProcess, subscribed } = preparePendingProcess();
118 const adapter = yield* PtyAdapter.PtyAdapter;
119 const fiber = yield* adapter.spawn(spawnInput).pipe(Effect.result, Effect.forkChild);
120 yield* subscribed;
121 if (failure === "exit") nativeProcess.events.emit("exit", { exitCode: 1 });
122 else if (failure === "error") nativeProcess._socket.emit("error", new Error("pipe failed"));
123 else nativeProcess._socket.emit(failure === "close" ? "close" : "ready_datapipe");
124 const result = yield* Fiber.join(fiber);
125 assert.equal(result._tag, "Failure");
126 if (result._tag === "Failure") assert.instanceOf(result.failure, PtyAdapter.PtySpawnError);
127 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
128 assert.equal(nativeProcess._socket.listenerCount("error"), 0);
129 assert.equal(nativeProcess._socket.listenerCount("close"), 0);
130 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
131 assert.equal(nativeProcess._agent.kill.mock.calls.length, 1);
132 }).pipe(Effect.provide(layerTest)),
133);
134
135it.effect("cancels the Windows connection without waiting for output", () =>
136 Effect.gen(function* () {
137 const { nativeProcess, subscribed } = preparePendingProcess();
138 const adapter = yield* PtyAdapter.PtyAdapter;
139 const fiber = yield* adapter.spawn(spawnInput).pipe(Effect.forkChild);
140 yield* subscribed;
141 yield* Fiber.interrupt(fiber);
142 assert.equal(nativeProcess._agent.kill.mock.calls.length, 1);
143 assert.equal(nativeProcess.kill.mock.calls.length, 0);
144 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
145 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
146 }).pipe(Effect.provide(layerTest)),
147);
148
149it.effect("reports an incompatible Windows readiness API instead of hanging", () =>
150 Effect.gen(function* () {
151 const nativeProcess = makeNativeProcess(0);
152 Reflect.deleteProperty(nativeProcess, "_socket");
153 spawn.mockReturnValueOnce(nativeProcess);
154 const adapter = yield* PtyAdapter.PtyAdapter;
155 const error = yield* adapter.spawn(spawnInput).pipe(Effect.flip);
156 assert.instanceOf(error, PtyAdapter.PtySpawnError);
157 assert.instanceOf(error.cause, Error);
158 assert.equal(error.cause.message, "Windows PTY readiness socket is unavailable.");
159 assert.equal(nativeProcess._agent.kill.mock.calls.length, 1);
160 }).pipe(Effect.provide(layerTest)),
161);
162
163it.effect.each(["win32", "linux", "darwin"] as const)(
164 "terminates through node-pty using %s semantics",
165 (platform) =>
166 Effect.gen(function* () {
167 const adapter = yield* PtyAdapter.PtyAdapter;
168 const process = yield* adapter.spawn({
169 shell: "test-shell",
170 cwd: ".",
171 cols: 80,
172 rows: 24,
173 env: {},
174 });
175 const nativeProcess = spawn.mock.results.at(-1)!.value;
176 nativeProcess.kill.mockImplementation((signal?: string) => {
177 if (platform === "win32" && signal) {
178 throw new Error("Signals not supported on windows.");
179 }
180 });
181
182 process.kill("SIGTERM");
183 process.kill("SIGKILL");
184 process.kill();
185
186 assert.deepEqual(
187 nativeProcess.kill.mock.calls,
188 platform === "win32"
189 ? [[undefined], [undefined], [undefined]]
190 : [["SIGTERM"], ["SIGKILL"], [undefined]],
191 );
192 }).pipe(Effect.provide(layerTestFor(platform))),
193);
194
195it.effect("spawns through the public adapter with the provided host references", () =>
196 Effect.gen(function* () {
197 spawn.mockClear();
198 const adapter = yield* PtyAdapter.PtyAdapter;
199 const process = yield* adapter.spawn({
200 shell: "powershell.exe",
201 args: ["-NoLogo"],
202 cwd: "C:\\workspace",
203 cols: 120,
204 rows: 40,
205 env: {},
206 });
207
208 assert.equal(process.pid, 42);
209 assert.equal(spawn.mock.calls.length, 1);
210 assert.deepEqual(spawn.mock.calls[0], [
211 "powershell.exe",
212 ["-NoLogo"],
213 {
214 cwd: "C:\\workspace",
215 cols: 120,
216 rows: 40,
217 env: { TERM: "xterm-256color" },
218 name: "xterm-256color",
219 },
220 ]);
221 }).pipe(Effect.provide(layerTest)),
222);
223
224it.effect("preserves a caller-provided TERM in the spawn env on win32", () =>
225 Effect.gen(function* () {
226 spawn.mockClear();
227 const adapter = yield* PtyAdapter.PtyAdapter;
228 yield* adapter.spawn({
229 shell: "powershell.exe",
230 cwd: "C:\\workspace",
231 cols: 80,
232 rows: 24,
233 env: { TERM: "xterm-direct" },
234 });
235
236 assert.equal(spawn.mock.calls.length, 1);
237 assert.deepEqual(spawn.mock.calls[0], [
238 "powershell.exe",
239 [],
240 {
241 cwd: "C:\\workspace",
242 cols: 80,
243 rows: 24,
244 env: { TERM: "xterm-direct" },
245 name: "xterm-256color",
246 },
247 ]);
248 }).pipe(Effect.provide(layerTest)),
249);
250
251it.effect("reports native module load failures as structured startup defects", () =>
252 Effect.gen(function* () {
253 const cause = new Error("native binding could not be loaded");
254 const exit = yield* NodePtyAdapter.make().pipe(
255 Effect.provideService(NodePtyAdapter.NodePtyModuleLoaderRef, () => Promise.reject(cause)),
256 Effect.exit,
257 );
258
259 assert.isTrue(Exit.isFailure(exit));
260 if (Exit.isFailure(exit)) {
261 assert.isTrue(Cause.hasDies(exit.cause));
262 const error = Cause.squash(exit.cause);
263 assert.instanceOf(error, NodePtyAdapter.NodePtyModuleLoadError);
264 assert.deepInclude(error, {
265 _tag: "NodePtyModuleLoadError",
266 platform: "win32",
267 architecture: "x64",
268 });
269 assert.equal(error.message, "Failed to load node-pty for win32-x64.");
270 }
271 }).pipe(
272 Effect.provide(
273 Layer.mergeAll(
274 NodeServices.layer,
275 Layer.succeed(HostProcess.Platform, "win32"),
276 Layer.succeed(HostProcess.Architecture, "x64"),
277 ),
278 ),
279 ),
280);
281
282it.effect.each([2048, 8])(
283 "preserves an exit during readiness handoff with scheduler budget %s",
284 (budget) =>
285 Effect.gen(function* () {
286 const { nativeProcess, subscribed } = preparePendingProcess();
287 const adapter = yield* PtyAdapter.PtyAdapter;
288 const exits: PtyAdapter.PtyExitEvent[] = [];
289 const fiber = yield* Effect.gen(function* () {
290 const process = yield* adapter.spawn(spawnInput);
291 process.onExit((event) => exits.push(event));
292 }).pipe(
293 Effect.provideService(Scheduler.MaxOpsBeforeYield, budget),
294 Effect.provideService(Scheduler.PreventSchedulerYield, false),
295 Effect.forkChild,
296 );
297 yield* subscribed;
298 nativeProcess.pid = 12345;
299 nativeProcess._socket.emit("ready_datapipe");
300 nativeProcess.events.emit("exit", { exitCode: 0 });
301 yield* Fiber.join(fiber);
302 assert.equal(exits.length, 1);
303 }).pipe(Effect.provide(layerTest)),
304);
305
306it.effect("replays an exit to late subscribers and respects unsubscription", () =>
307 Effect.gen(function* () {
308 const adapter = yield* PtyAdapter.PtyAdapter;
309 const process = yield* adapter.spawn(spawnInput);
310 const nativeProcess = spawn.mock.results.at(-1)!.value;
311 const removed = vi.fn();
312 process.onExit(removed)();
313 nativeProcess.events.emit("exit", { exitCode: 7, signal: 2 });
314 const late = vi.fn();
315 process.onExit(late);
316 nativeProcess.events.emit("exit", { exitCode: 9 });
317 assert.equal(removed.mock.calls.length, 0);
318 assert.deepEqual(late.mock.calls, [[{ exitCode: 7, signal: 2 }]]);
319 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
320 }).pipe(Effect.provide(layerTest)),
321);
322
323it.effect.each(["spawn", "interrupt"] as const)(
324 "logs cleanup failures without replacing %s",
325 (failure) =>
326 Effect.gen(function* () {
327 const { nativeProcess, subscribed } = preparePendingProcess();
328 const killError = new Error("native kill failed");
329 nativeProcess._agent.kill.mockImplementation(() => {
330 throw killError;
331 });
332 const messages: unknown[] = [];
333 const logger = Logger.make(({ message }) => {
334 messages.push(message);
335 });
336 const adapter = yield* PtyAdapter.PtyAdapter;
337 const fiber = yield* adapter
338 .spawn(spawnInput)
339 .pipe(
340 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
341 Effect.forkChild,
342 );
343 yield* subscribed;
344 const spawnError = new Error("pipe failed");
345 if (failure === "interrupt") yield* Fiber.interrupt(fiber);
346 else nativeProcess._socket.emit("error", spawnError);
347 const exit = yield* Fiber.await(fiber);
348 assert.isTrue(Exit.isFailure(exit));
349 if (Exit.isFailure(exit)) {
350 if (failure === "interrupt") assert.isTrue(Cause.hasInterrupts(exit.cause));
351 else {
352 const error = Cause.squash(exit.cause);
353 assert.instanceOf(error, PtyAdapter.PtySpawnError);
354 assert.equal(error.cause, spawnError);
355 }
356 }
357 assert.equal(messages.length, 1);
358 expect(messages[0]).toMatchObject([
359 "failed to cancel Windows terminal startup",
360 { terminalPid: 0, cause: { cause: killError } },
361 ]);
362 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
363 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
364 }).pipe(Effect.provide(layerTest)),
365);