| 1 | import * as NodeEvents from "node:events"; |
| 2 | import * as NodeNet from "node:net"; |
| 3 | |
| 4 | import * as NodeServices from "@effect/platform-node/NodeServices"; |
| 5 | import { assert, it } from "@effect/vitest"; |
| 6 | import * as HostProcess from "@t3tools/shared/HostProcess"; |
| 7 | import * as Cause from "effect/Cause"; |
| 8 | import * as Effect from "effect/Effect"; |
| 9 | import * as Exit from "effect/Exit"; |
| 10 | import * as Fiber from "effect/Fiber"; |
| 11 | import * as Layer from "effect/Layer"; |
| 12 | import * as Logger from "effect/Logger"; |
| 13 | import * as Scheduler from "effect/Scheduler"; |
| 14 | import { expect, vi } from "vite-plus/test"; |
| 15 | |
| 16 | import * as NodePtyAdapter from "./NodePtyAdapter.ts"; |
| 17 | import * as PtyAdapter from "@t3tools/shared/PtyAdapter"; |
| 18 | |
| 19 | function 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 | |
| 48 | const spawn = vi.fn(() => makeNativeProcess()); |
| 49 | |
| 50 | function 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 | |
| 60 | const spawnInput = { shell: "powershell.exe", cwd: ".", cols: 80, rows: 24, env: {} }; |
| 61 | |
| 62 | const fakeNodePty = { spawn } as unknown as typeof import("node-pty"); |
| 63 | |
| 64 | const 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 | |
| 76 | const layerTest = layerTestFor(); |
| 77 | |
| 78 | it.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 | |
| 113 | it.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 | |
| 135 | it.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 | |
| 149 | it.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 | |
| 163 | it.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 | |
| 195 | it.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 | |
| 224 | it.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 | |
| 251 | it.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 | |
| 282 | it.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 | |
| 306 | it.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 | |
| 323 | it.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 | ); |