| author | |
| committer | |
| log | b528a701102f95089544596856a3f208f4404613 |
| tree | 7dace126f6357e9444e45d41a07408eab5fb4e69 |
| parent | 72330e22c0d08159641ab10579eb75393c4d95b6 |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: UtkarshUsername <putkarsh184@gmail.com>
Co-authored-by: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com>4 files changed, 396 insertions(+), 34 deletions(-)
apps/server/src/terminal/Manager.test.ts+34| ... | @@ -56,6 +56,7 @@ class FakePtyProcess implements PtyAdapter.PtyProcess { | ... | @@ -56,6 +56,7 @@ class FakePtyProcess implements PtyAdapter.PtyProcess { |
| 56 | private readonly dataListeners = new Set<(data: string) => void>(); | 56 | private readonly dataListeners = new Set<(data: string) => void>(); |
| 57 | private readonly exitListeners = new Set<(event: PtyAdapter.PtyExitEvent) => void>(); | 57 | private readonly exitListeners = new Set<(event: PtyAdapter.PtyExitEvent) => void>(); |
| 58 | killed = false; | 58 | killed = false; |
| 59 | exitOnSubscribe: PtyAdapter.PtyExitEvent | undefined; | ||
| 59 | 60 | ||
| 60 | constructor(pid: number) { | 61 | constructor(pid: number) { |
| 61 | this.pid = pid; | 62 | this.pid = pid; |
| ... | @@ -88,6 +89,7 @@ class FakePtyProcess implements PtyAdapter.PtyProcess { | ... | @@ -88,6 +89,7 @@ class FakePtyProcess implements PtyAdapter.PtyProcess { |
| 88 | } | 89 | } |
| 89 | 90 | ||
| 90 | onExit(callback: (event: PtyAdapter.PtyExitEvent) => void): () => void { | 91 | onExit(callback: (event: PtyAdapter.PtyExitEvent) => void): () => void { |
| 92 | if (this.exitOnSubscribe) callback(this.exitOnSubscribe); | ||
| 91 | this.exitListeners.add(callback); | 93 | this.exitListeners.add(callback); |
| 92 | return () => { | 94 | return () => { |
| 93 | this.exitListeners.delete(callback); | 95 | this.exitListeners.delete(callback); |
| ... | @@ -113,6 +115,7 @@ class FakePtyAdapter { | ... | @@ -113,6 +115,7 @@ class FakePtyAdapter { |
| 113 | readonly spawnFailures: Error[] = []; | 115 | readonly spawnFailures: Error[] = []; |
| 114 | private readonly mode: "sync" | "async"; | 116 | private readonly mode: "sync" | "async"; |
| 115 | private nextPid = 9000; | 117 | private nextPid = 9000; |
| 118 | exitOnSubscribe: PtyAdapter.PtyExitEvent | undefined; | ||
| 116 | 119 | ||
| 117 | constructor(mode: "sync" | "async" = "sync") { | 120 | constructor(mode: "sync" | "async" = "sync") { |
| 118 | this.mode = mode; | 121 | this.mode = mode; |
| ... | @@ -133,6 +136,7 @@ class FakePtyAdapter { | ... | @@ -133,6 +136,7 @@ class FakePtyAdapter { |
| 133 | ); | 136 | ); |
| 134 | } | 137 | } |
| 135 | const process = new FakePtyProcess(this.nextPid++); | 138 | const process = new FakePtyProcess(this.nextPid++); |
| 139 | process.exitOnSubscribe = this.exitOnSubscribe; | ||
| 136 | this.processes.push(process); | 140 | this.processes.push(process); |
| 137 | if (this.mode === "async") { | 141 | if (this.mode === "async") { |
| 138 | return Effect.tryPromise({ | 142 | return Effect.tryPromise({ |
| ... | @@ -628,6 +632,36 @@ it.layer( | ... | @@ -628,6 +632,36 @@ it.layer( |
| 628 | }), | 632 | }), |
| 629 | ); | 633 | ); |
| 630 | 634 | ||
| 635 | it.effect("handles an exit replayed during subscription after publishing startup", () => | ||
| 636 | Effect.gen(function* () { | ||
| 637 | const ptyAdapter = new FakePtyAdapter(); | ||
| 638 | ptyAdapter.exitOnSubscribe = { exitCode: 7, signal: null }; | ||
| 639 | const { manager, getEvents } = yield* createManager(5, { ptyAdapter }); | ||
| 640 | const exited = yield* Deferred.make<void>(); | ||
| 641 | const unsubscribe = yield* manager.subscribe((event) => | ||
| 642 | event.type === "exited" | ||
| 643 | ? Deferred.succeed(exited, undefined).pipe(Effect.asVoid) | ||
| 644 | : Effect.void, | ||
| 645 | ); | ||
| 646 | yield* Effect.addFinalizer(() => Effect.sync(unsubscribe)); | ||
| 647 | yield* manager.open(openInput()); | ||
| 648 | yield* Deferred.await(exited); | ||
| 649 | const events = yield* getEvents; | ||
| 650 | expect(events.map((event) => event.type)).toEqual(["started", "exited"]); | ||
| 651 | expect(events[1]).toMatchObject({ exitCode: 7 }); | ||
| 652 | const attached: TerminalAttachStreamEvent[] = []; | ||
| 653 | const stopAttach = yield* manager.attachStream(openInput(), (event) => | ||
| 654 | Effect.sync(() => { | ||
| 655 | attached.push(event); | ||
| 656 | }), | ||
| 657 | ); | ||
| 658 | yield* Effect.addFinalizer(() => Effect.sync(stopAttach)); | ||
| 659 | expect(attached.find((event) => event.type === "snapshot")).toMatchObject({ | ||
| 660 | snapshot: { status: "exited", exitCode: 7 }, | ||
| 661 | }); | ||
| 662 | }), | ||
| 663 | ); | ||
| 664 | |||
| 631 | it.effect("supports asynchronous PTY spawn effects", () => | 665 | it.effect("supports asynchronous PTY spawn effects", () => |
| 632 | Effect.gen(function* () { | 666 | Effect.gen(function* () { |
| 633 | const { manager, ptyAdapter } = yield* createManager(5, { | 667 | const { manager, ptyAdapter } = yield* createManager(5, { |
apps/server/src/terminal/Manager.ts+17-14| ... | @@ -2243,18 +2243,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func | ... | @@ -2243,18 +2243,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func |
| 2243 | startedShell = spawnResult.shellLabel; | 2243 | startedShell = spawnResult.shellLabel; |
| 2244 | 2244 | ||
| 2245 | const processPid = ptyProcess.pid; | 2245 | const processPid = ptyProcess.pid; |
| 2246 | const unsubscribeData = ptyProcess.onData((data) => { | 2246 | let eventsActivated = false; |
| 2247 | if (!enqueueProcessEvent(session, processPid, { type: "output", data })) { | ||
| 2248 | return; | ||
| 2249 | } | ||
| 2250 | runFork(drainProcessEvents(session, processPid)); | ||
| 2251 | }); | ||
| 2252 | const unsubscribeExit = ptyProcess.onExit((event) => { | ||
| 2253 | if (!enqueueProcessEvent(session, processPid, { type: "exit", event })) { | ||
| 2254 | return; | ||
| 2255 | } | ||
| 2256 | runFork(drainProcessEvents(session, processPid)); | ||
| 2257 | }); | ||
| 2258 | 2247 | ||
| 2259 | let eventStamp: ReturnType<typeof advanceEventSequence> = { | 2248 | let eventStamp: ReturnType<typeof advanceEventSequence> = { |
| 2260 | updatedAt: session.updatedAt, | 2249 | updatedAt: session.updatedAt, |
| ... | @@ -2264,8 +2253,19 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func | ... | @@ -2264,8 +2253,19 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func |
| 2264 | session.process = ptyProcess; | 2253 | session.process = ptyProcess; |
| 2265 | session.pid = processPid; | 2254 | session.pid = processPid; |
| 2266 | session.status = "running"; | 2255 | session.status = "running"; |
| 2267 | session.unsubscribeData = unsubscribeData; | 2256 | // onExit may replay an exit immediately; accept it before subscribing. |
| 2268 | session.unsubscribeExit = unsubscribeExit; | 2257 | session.unsubscribeData = spawnResult.process.onData((data) => { |
| 2258 | if (!enqueueProcessEvent(session, processPid, { type: "output", data })) { | ||
| 2259 | return; | ||
| 2260 | } | ||
| 2261 | if (eventsActivated) runFork(drainProcessEvents(session, processPid)); | ||
| 2262 | }); | ||
| 2263 | session.unsubscribeExit = spawnResult.process.onExit((event) => { | ||
| 2264 | if (!enqueueProcessEvent(session, processPid, { type: "exit", event })) { | ||
| 2265 | return; | ||
| 2266 | } | ||
| 2267 | if (eventsActivated) runFork(drainProcessEvents(session, processPid)); | ||
| 2268 | }); | ||
| 2269 | eventStamp = advanceEventSequence(session); | 2269 | eventStamp = advanceEventSequence(session); |
| 2270 | return [undefined, state] as const; | 2270 | return [undefined, state] as const; |
| 2271 | }); | 2271 | }); |
| ... | @@ -2277,6 +2277,9 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func | ... | @@ -2277,6 +2277,9 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func |
| 2277 | sequence: eventStamp.sequence, | 2277 | sequence: eventStamp.sequence, |
| 2278 | snapshot: snapshot(session), | 2278 | snapshot: snapshot(session), |
| 2279 | }); | 2279 | }); |
| 2280 | // Publish startup before draining any events replayed during subscription. | ||
| 2281 | eventsActivated = true; | ||
| 2282 | if (session.processEventDrainRunning) runFork(drainProcessEvents(session, processPid)); | ||
| 2280 | }), | 2283 | }), |
| 2281 | ), | 2284 | ), |
| 2282 | ), | 2285 | ), |
apps/server/src/terminal/NodePtyAdapter.test.ts+219-9| ... | @@ -1,23 +1,63 @@ | ... | @@ -1,23 +1,63 @@ |
| 1 | import * as NodeEvents from "node:events"; | ||
| 2 | import * as NodeNet from "node:net"; | ||
| 3 | |||
| 1 | import * as NodeServices from "@effect/platform-node/NodeServices"; | 4 | import * as NodeServices from "@effect/platform-node/NodeServices"; |
| 2 | import { assert, it } from "@effect/vitest"; | 5 | import { assert, it } from "@effect/vitest"; |
| 3 | import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess"; | 6 | import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess"; |
| 4 | import * as Cause from "effect/Cause"; | 7 | import * as Cause from "effect/Cause"; |
| 5 | import * as Effect from "effect/Effect"; | 8 | import * as Effect from "effect/Effect"; |
| 6 | import * as Exit from "effect/Exit"; | 9 | import * as Exit from "effect/Exit"; |
| 10 | import * as Fiber from "effect/Fiber"; | ||
| 7 | import * as Layer from "effect/Layer"; | 11 | import * as Layer from "effect/Layer"; |
| 8 | import { vi } from "vite-plus/test"; | 12 | import * as Logger from "effect/Logger"; |
| 13 | import * as Scheduler from "effect/Scheduler"; | ||
| 14 | import { expect, vi } from "vite-plus/test"; | ||
| 9 | 15 | ||
| 10 | import * as NodePtyAdapter from "./NodePtyAdapter.ts"; | 16 | import * as NodePtyAdapter from "./NodePtyAdapter.ts"; |
| 11 | import * as PtyAdapter from "./PtyAdapter.ts"; | 17 | import * as PtyAdapter from "./PtyAdapter.ts"; |
| 12 | 18 | ||
| 13 | const spawn = vi.fn(() => ({ | 19 | function makeNativeProcess(pid = 42) { |
| 14 | pid: 42, | 20 | const events = new NodeEvents.EventEmitter(); |
| 15 | write: vi.fn(), | 21 | return { |
| 16 | resize: vi.fn(), | 22 | pid, |
| 17 | kill: vi.fn(), | 23 | _socket: new NodeNet.Socket(), |
| 18 | onData: vi.fn(() => ({ dispose: vi.fn() })), | 24 | _agent: { kill: vi.fn() }, |
| 19 | onExit: vi.fn(() => ({ dispose: vi.fn() })), | 25 | write: vi.fn(), |
| 20 | })); | 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: {} }; | ||
| 21 | 61 | ||
| 22 | const fakeNodePty = { spawn } as unknown as typeof import("node-pty"); | 62 | const fakeNodePty = { spawn } as unknown as typeof import("node-pty"); |
| 23 | 63 | ||
| ... | @@ -35,6 +75,91 @@ const makeTestLayer = (platform: NodeJS.Platform = "win32") => | ... | @@ -35,6 +75,91 @@ const makeTestLayer = (platform: NodeJS.Platform = "win32") => |
| 35 | 75 | ||
| 36 | const testLayer = makeTestLayer(); | 76 | const testLayer = makeTestLayer(); |
| 37 | 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(testLayer)), | ||
| 111 | ); | ||
| 112 | |||
| 113 | for (const failure of ["exit", "close", "error", "invalid-pid"] as const) { | ||
| 114 | it.effect(`fails Windows startup on ${failure} and cleans up`, () => | ||
| 115 | Effect.gen(function* () { | ||
| 116 | const { nativeProcess, subscribed } = preparePendingProcess(); | ||
| 117 | const adapter = yield* PtyAdapter.PtyAdapter; | ||
| 118 | const fiber = yield* adapter.spawn(spawnInput).pipe(Effect.result, Effect.forkChild); | ||
| 119 | yield* subscribed; | ||
| 120 | if (failure === "exit") nativeProcess.events.emit("exit", { exitCode: 1 }); | ||
| 121 | else if (failure === "error") nativeProcess._socket.emit("error", new Error("pipe failed")); | ||
| 122 | else nativeProcess._socket.emit(failure === "close" ? "close" : "ready_datapipe"); | ||
| 123 | const result = yield* Fiber.join(fiber); | ||
| 124 | assert.equal(result._tag, "Failure"); | ||
| 125 | if (result._tag === "Failure") assert.instanceOf(result.failure, PtyAdapter.PtySpawnError); | ||
| 126 | assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0); | ||
| 127 | assert.equal(nativeProcess._socket.listenerCount("error"), 0); | ||
| 128 | assert.equal(nativeProcess._socket.listenerCount("close"), 0); | ||
| 129 | assert.equal(nativeProcess.events.listenerCount("exit"), 0); | ||
| 130 | assert.equal(nativeProcess._agent.kill.mock.calls.length, 1); | ||
| 131 | }).pipe(Effect.provide(testLayer)), | ||
| 132 | ); | ||
| 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(testLayer)), | ||
| 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(testLayer)), | ||
| 161 | ); | ||
| 162 | |||
| 38 | for (const platform of ["win32", "linux", "darwin"] as const) { | 163 | for (const platform of ["win32", "linux", "darwin"] as const) { |
| 39 | it.effect(`terminates through node-pty using ${platform} semantics`, () => | 164 | it.effect(`terminates through node-pty using ${platform} semantics`, () => |
| 40 | Effect.gen(function* () { | 165 | Effect.gen(function* () { |
| ... | @@ -153,3 +278,88 @@ it.effect("reports native module load failures as structured startup defects", ( | ... | @@ -153,3 +278,88 @@ it.effect("reports native module load failures as structured startup defects", ( |
| 153 | ), | 278 | ), |
| 154 | ), | 279 | ), |
| 155 | ); | 280 | ); |
| 281 | |||
| 282 | for (const budget of [2048, 8]) { | ||
| 283 | it.effect(`preserves an exit during readiness handoff with scheduler budget ${budget}`, () => | ||
| 284 | Effect.gen(function* () { | ||
| 285 | const { nativeProcess, subscribed } = preparePendingProcess(); | ||
| 286 | const adapter = yield* PtyAdapter.PtyAdapter; | ||
| 287 | const exits: PtyAdapter.PtyExitEvent[] = []; | ||
| 288 | const fiber = yield* Effect.gen(function* () { | ||
| 289 | const process = yield* adapter.spawn(spawnInput); | ||
| 290 | process.onExit((event) => exits.push(event)); | ||
| 291 | }).pipe( | ||
| 292 | Effect.provideService(Scheduler.MaxOpsBeforeYield, budget), | ||
| 293 | Effect.provideService(Scheduler.PreventSchedulerYield, false), | ||
| 294 | Effect.forkChild, | ||
| 295 | ); | ||
| 296 | yield* subscribed; | ||
| 297 | nativeProcess.pid = 12345; | ||
| 298 | nativeProcess._socket.emit("ready_datapipe"); | ||
| 299 | nativeProcess.events.emit("exit", { exitCode: 0 }); | ||
| 300 | yield* Fiber.join(fiber); | ||
| 301 | assert.equal(exits.length, 1); | ||
| 302 | }).pipe(Effect.provide(testLayer)), | ||
| 303 | ); | ||
| 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(testLayer)), | ||
| 321 | ); | ||
| 322 | |||
| 323 | for (const failure of ["spawn", "interrupt"] as const) { | ||
| 324 | it.effect(`logs cleanup failures without replacing ${failure}`, () => | ||
| 325 | Effect.gen(function* () { | ||
| 326 | const { nativeProcess, subscribed } = preparePendingProcess(); | ||
| 327 | const killError = new Error("native kill failed"); | ||
| 328 | nativeProcess._agent.kill.mockImplementation(() => { | ||
| 329 | throw killError; | ||
| 330 | }); | ||
| 331 | const messages: unknown[] = []; | ||
| 332 | const logger = Logger.make(({ message }) => { | ||
| 333 | messages.push(message); | ||
| 334 | }); | ||
| 335 | const adapter = yield* PtyAdapter.PtyAdapter; | ||
| 336 | const fiber = yield* adapter | ||
| 337 | .spawn(spawnInput) | ||
| 338 | .pipe( | ||
| 339 | Effect.provide(Logger.layer([logger], { mergeWithExisting: false })), | ||
| 340 | Effect.forkChild, | ||
| 341 | ); | ||
| 342 | yield* subscribed; | ||
| 343 | const spawnError = new Error("pipe failed"); | ||
| 344 | if (failure === "interrupt") yield* Fiber.interrupt(fiber); | ||
| 345 | else nativeProcess._socket.emit("error", spawnError); | ||
| 346 | const exit = yield* Fiber.await(fiber); | ||
| 347 | assert.isTrue(Exit.isFailure(exit)); | ||
| 348 | if (Exit.isFailure(exit)) { | ||
| 349 | if (failure === "interrupt") assert.isTrue(Cause.hasInterrupts(exit.cause)); | ||
| 350 | else { | ||
| 351 | const error = Cause.squash(exit.cause); | ||
| 352 | assert.instanceOf(error, PtyAdapter.PtySpawnError); | ||
| 353 | assert.equal(error.cause, spawnError); | ||
| 354 | } | ||
| 355 | } | ||
| 356 | assert.equal(messages.length, 1); | ||
| 357 | expect(messages[0]).toMatchObject([ | ||
| 358 | "failed to cancel Windows terminal startup", | ||
| 359 | { terminalPid: 0, cause: { cause: killError } }, | ||
| 360 | ]); | ||
| 361 | assert.equal(nativeProcess.events.listenerCount("exit"), 0); | ||
| 362 | assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0); | ||
| 363 | }).pipe(Effect.provide(testLayer)), | ||
| 364 | ); | ||
| 365 | } |
apps/server/src/terminal/NodePtyAdapter.ts+126-11| ... | @@ -1,4 +1,5 @@ | ... | @@ -1,4 +1,5 @@ |
| 1 | import * as NodeModule from "node:module"; | 1 | import * as NodeModule from "node:module"; |
| 2 | import * as NodeNet from "node:net"; | ||
| 2 | 3 | ||
| 3 | import * as Context from "effect/Context"; | 4 | import * as Context from "effect/Context"; |
| 4 | import * as Effect from "effect/Effect"; | 5 | import * as Effect from "effect/Effect"; |
| ... | @@ -82,13 +83,112 @@ const ensureNodePtySpawnHelperExecutable = Effect.fn(function* () { | ... | @@ -82,13 +83,112 @@ const ensureNodePtySpawnHelperExecutable = Effect.fn(function* () { |
| 82 | yield* fs.chmod(helperPath, 0o755).pipe(Effect.orElseSucceed(() => undefined)); | 83 | yield* fs.chmod(helperPath, 0o755).pipe(Effect.orElseSucceed(() => undefined)); |
| 83 | }); | 84 | }); |
| 84 | 85 | ||
| 86 | /** | ||
| 87 | * Waits for Windows process creation so the manager receives a valid PID. | ||
| 88 | * node-pty defers creation to avoid blocking on named pipes: | ||
| 89 | * https://github.com/microsoft/node-pty/pull/885 | ||
| 90 | * T3 adopted that behavior when upgrading from 1.1.0 to 1.2.0-beta.15: | ||
| 91 | * https://github.com/pingdotgg/t3code/pull/13748 | ||
| 92 | * Its public API has no readiness event. The private ready_datapipe handler sets | ||
| 93 | * pid before our listener runs. | ||
| 94 | */ | ||
| 95 | const waitForWindowsPid = ( | ||
| 96 | process: import("node-pty").IPty, | ||
| 97 | trackedProcess: NodePtyProcess, | ||
| 98 | shell: string, | ||
| 99 | ) => | ||
| 100 | Effect.callback<void, PtyAdapter.PtySpawnError>((resume) => { | ||
| 101 | const hasPid = () => Number.isInteger(process.pid) && process.pid > 0; | ||
| 102 | const failure = (cause: unknown) => | ||
| 103 | Effect.fail(new PtyAdapter.PtySpawnError({ adapter: "node-pty", shell, cause })); | ||
| 104 | |||
| 105 | if (hasPid()) { | ||
| 106 | resume(Effect.void); | ||
| 107 | return; | ||
| 108 | } | ||
| 109 | |||
| 110 | if (!("_socket" in process) || !(process._socket instanceof NodeNet.Socket)) { | ||
| 111 | resume(failure(new Error("Windows PTY readiness socket is unavailable."))); | ||
| 112 | return; | ||
| 113 | } | ||
| 114 | |||
| 115 | const socket = process._socket; | ||
| 116 | const onReady = () => { | ||
| 117 | cleanup(); | ||
| 118 | resume( | ||
| 119 | hasPid() | ||
| 120 | ? Effect.void | ||
| 121 | : failure(new Error("Windows PTY became ready without a valid PID.")), | ||
| 122 | ); | ||
| 123 | }; | ||
| 124 | const onError = (cause: Error) => { | ||
| 125 | cleanup(); | ||
| 126 | resume(failure(cause)); | ||
| 127 | }; | ||
| 128 | const onClose = () => onError(new Error("Windows PTY closed before its PID was available.")); | ||
| 129 | let stopExit = () => {}; | ||
| 130 | const cleanup = () => { | ||
| 131 | socket.off("ready_datapipe", onReady); | ||
| 132 | socket.off("error", onError); | ||
| 133 | socket.off("close", onClose); | ||
| 134 | stopExit(); | ||
| 135 | }; | ||
| 136 | socket.once("ready_datapipe", onReady); | ||
| 137 | socket.once("error", onError); | ||
| 138 | socket.once("close", onClose); | ||
| 139 | stopExit = trackedProcess.onExit(({ exitCode }) => | ||
| 140 | onError( | ||
| 141 | new Error(`Windows PTY exited before its PID was available (exit code ${exitCode}).`), | ||
| 142 | ), | ||
| 143 | ); | ||
| 144 | return Effect.sync(cleanup); | ||
| 145 | }); | ||
| 146 | |||
| 147 | /** | ||
| 148 | * Cancels Windows startup without waiting for the first output, unlike public kill(). | ||
| 149 | * The private agent can cancel the pending connection before a child exists. | ||
| 150 | * Cleanup failures are logged without replacing the startup failure. | ||
| 151 | */ | ||
| 152 | const killStartingWindowsPty = (process: import("node-pty").IPty) => | ||
| 153 | Effect.try(() => { | ||
| 154 | if ( | ||
| 155 | "_agent" in process && | ||
| 156 | typeof process._agent === "object" && | ||
| 157 | process._agent !== null && | ||
| 158 | "kill" in process._agent && | ||
| 159 | typeof process._agent.kill === "function" | ||
| 160 | ) { | ||
| 161 | process._agent.kill(); | ||
| 162 | } else { | ||
| 163 | process.kill(); | ||
| 164 | } | ||
| 165 | }).pipe( | ||
| 166 | Effect.catch((error) => | ||
| 167 | Effect.logWarning("failed to cancel Windows terminal startup", { | ||
| 168 | terminalPid: process.pid, | ||
| 169 | cause: error, | ||
| 170 | }), | ||
| 171 | ), | ||
| 172 | ); | ||
| 173 | |||
| 85 | class NodePtyProcess implements PtyAdapter.PtyProcess { | 174 | class NodePtyProcess implements PtyAdapter.PtyProcess { |
| 86 | private readonly process: import("node-pty").IPty; | 175 | private readonly process: import("node-pty").IPty; |
| 87 | private readonly platform: NodeJS.Platform; | 176 | private readonly platform: NodeJS.Platform; |
| 177 | private exitEvent: PtyAdapter.PtyExitEvent | undefined; | ||
| 178 | private readonly exitListeners = new Set<(event: PtyAdapter.PtyExitEvent) => void>(); | ||
| 179 | private readonly exitSubscription: import("node-pty").IDisposable; | ||
| 88 | 180 | ||
| 89 | constructor(process: import("node-pty").IPty, platform: NodeJS.Platform) { | 181 | constructor(process: import("node-pty").IPty, platform: NodeJS.Platform) { |
| 90 | this.process = process; | 182 | this.process = process; |
| 91 | this.platform = platform; | 183 | this.platform = platform; |
| 184 | // Retain exits while Windows readiness and the manager hand off the process. | ||
| 185 | this.exitSubscription = process.onExit((event) => { | ||
| 186 | if (this.exitEvent) return; | ||
| 187 | this.exitEvent = { exitCode: event.exitCode, signal: event.signal ?? null }; | ||
| 188 | this.exitSubscription.dispose(); | ||
| 189 | for (const listener of this.exitListeners) listener(this.exitEvent); | ||
| 190 | this.exitListeners.clear(); | ||
| 191 | }); | ||
| 92 | } | 192 | } |
| 93 | 193 | ||
| 94 | get pid(): number { | 194 | get pid(): number { |
| ... | @@ -116,16 +216,20 @@ class NodePtyProcess implements PtyAdapter.PtyProcess { | ... | @@ -116,16 +216,20 @@ class NodePtyProcess implements PtyAdapter.PtyProcess { |
| 116 | } | 216 | } |
| 117 | 217 | ||
| 118 | onExit(callback: (event: PtyAdapter.PtyExitEvent) => void): () => void { | 218 | onExit(callback: (event: PtyAdapter.PtyExitEvent) => void): () => void { |
| 119 | const disposable = this.process.onExit((event) => { | 219 | if (this.exitEvent) { |
| 120 | callback({ | 220 | callback(this.exitEvent); |
| 121 | exitCode: event.exitCode, | 221 | return () => {}; |
| 122 | signal: event.signal ?? null, | 222 | } |
| 123 | }); | 223 | this.exitListeners.add(callback); |
| 124 | }); | ||
| 125 | return () => { | 224 | return () => { |
| 126 | disposable.dispose(); | 225 | this.exitListeners.delete(callback); |
| 127 | }; | 226 | }; |
| 128 | } | 227 | } |
| 228 | |||
| 229 | disposeExitSubscription(): void { | ||
| 230 | this.exitSubscription.dispose(); | ||
| 231 | this.exitListeners.clear(); | ||
| 232 | } | ||
| 129 | } | 233 | } |
| 130 | 234 | ||
| 131 | export const make = Effect.fn("NodePtyAdapter.make")(function* () { | 235 | export const make = Effect.fn("NodePtyAdapter.make")(function* () { |
| ... | @@ -166,14 +270,16 @@ export const make = Effect.fn("NodePtyAdapter.make")(function* () { | ... | @@ -166,14 +270,16 @@ export const make = Effect.fn("NodePtyAdapter.make")(function* () { |
| 166 | ? { ...input.env, TERM: "xterm-256color" } | 270 | ? { ...input.env, TERM: "xterm-256color" } |
| 167 | : input.env; | 271 | : input.env; |
| 168 | const ptyProcess = yield* Effect.try({ | 272 | const ptyProcess = yield* Effect.try({ |
| 169 | try: () => | 273 | try: () => { |
| 170 | nodePty.spawn(input.shell, input.args ?? [], { | 274 | const nativeProcess = nodePty.spawn(input.shell, input.args ?? [], { |
| 171 | cwd: input.cwd, | 275 | cwd: input.cwd, |
| 172 | cols: input.cols, | 276 | cols: input.cols, |
| 173 | rows: input.rows, | 277 | rows: input.rows, |
| 174 | env, | 278 | env, |
| 175 | name: "xterm-256color", | 279 | name: "xterm-256color", |
| 176 | }), | 280 | }); |
| 281 | return { nativeProcess, process: new NodePtyProcess(nativeProcess, platform) }; | ||
| 282 | }, | ||
| 177 | catch: (cause) => | 283 | catch: (cause) => |
| 178 | new PtyAdapter.PtySpawnError({ | 284 | new PtyAdapter.PtySpawnError({ |
| 179 | adapter: "node-pty", | 285 | adapter: "node-pty", |
| ... | @@ -181,7 +287,16 @@ export const make = Effect.fn("NodePtyAdapter.make")(function* () { | ... | @@ -181,7 +287,16 @@ export const make = Effect.fn("NodePtyAdapter.make")(function* () { |
| 181 | cause, | 287 | cause, |
| 182 | }), | 288 | }), |
| 183 | }); | 289 | }); |
| 184 | return new NodePtyProcess(ptyProcess, platform); | 290 | if (platform === "win32") { |
| 291 | yield* waitForWindowsPid(ptyProcess.nativeProcess, ptyProcess.process, input.shell).pipe( | ||
| 292 | Effect.onError(() => | ||
| 293 | Effect.sync(() => ptyProcess.process.disposeExitSubscription()).pipe( | ||
| 294 | Effect.andThen(killStartingWindowsPty(ptyProcess.nativeProcess)), | ||
| 295 | ), | ||
| 296 | ), | ||
| 297 | ); | ||
| 298 | } | ||
| 299 | return ptyProcess.process; | ||
| 185 | }), | 300 | }), |
| 186 | }); | 301 | }); |
| 187 | }); | 302 | }); |