diff --git a/paseo-omp/server/provider/mcp-transport.ts b/paseo-omp/server/provider/mcp-transport.ts index efd73407..00ffd997 100644 --- a/paseo-omp/server/provider/mcp-transport.ts +++ b/paseo-omp/server/provider/mcp-transport.ts @@ -225,7 +225,7 @@ export class SupervisedStdioClientTransport implements Transport { return; } const treeCleanup = this.startTreeCleanup(); - if (!this.exited) { + if (this.platform !== "win32" && !this.exited) { try { child.stdin.end(); } catch { @@ -233,6 +233,13 @@ export class SupervisedStdioClientTransport implements Transport { } } const terminated = await treeCleanup; + if (this.platform === "win32" && terminated && !this.exited) { + try { + child.stdin.end(); + } catch { + // Process-tree cleanup remains authoritative when the input channel is already closed. + } + } const exited = this.spawnFailedWithoutProcess || this.exited || diff --git a/paseo-omp/server/provider/omp-rpc-transport.ts b/paseo-omp/server/provider/omp-rpc-transport.ts index e48e02b2..d44813aa 100644 --- a/paseo-omp/server/provider/omp-rpc-transport.ts +++ b/paseo-omp/server/provider/omp-rpc-transport.ts @@ -475,7 +475,7 @@ export class OmpRpcProcess { this.clearChunk(); this.failPending(new Error("OMP RPC process was closed")); const cleanupPromise = this.startTreeCleanup(); - if (!this.exited) { + if (process.platform !== "win32" && !this.exited) { try { this.child.stdin.end(); } catch { @@ -484,6 +484,13 @@ export class OmpRpcProcess { } const cleanup = await cleanupPromise; if (cleanup !== "verified") throw new Error("OMP RPC process tree cleanup failed"); + if (process.platform === "win32" && !this.exited) { + try { + this.child.stdin.end(); + } catch { + // Process-tree cleanup remains authoritative when the input channel is already closed. + } + } if ( !this.spawnFailedWithoutProcess && !this.exited && diff --git a/paseo-omp/tests/mcp-transport.test.ts b/paseo-omp/tests/mcp-transport.test.ts index c0e9d36b..4b2c39a5 100644 --- a/paseo-omp/tests/mcp-transport.test.ts +++ b/paseo-omp/tests/mcp-transport.test.ts @@ -155,28 +155,50 @@ describe("MCP transport boundaries", () => { await expect(closing).rejects.toThrow("transport cleanup failed"); }); - test("starts stdio tree cleanup before stdin shutdown can release the Windows tree root", async () => { + test("keeps stdin open until Windows process-tree cleanup settles", async () => { const child = new FakeMcpChild(); - let stdinEndedAtCleanup: boolean | undefined; + const treeCleanup = Promise.withResolvers(); const transport = new SupervisedStdioClientTransport( { command: "mcp-server", cwd: "C:\\workspace" }, { platform: "win32", spawnProcess: () => child.asChildProcess(), - terminateProcessTree: async () => { - stdinEndedAtCleanup = child.stdin.writableEnded; - queueMicrotask(() => child.emit("exit", 0, null)); - return true; - }, + terminateProcessTree: async () => await treeCleanup.promise, }, ); const starting = transport.start(); child.emit("spawn"); await starting; - await transport.close(); + const closing = transport.close(); + expect(child.stdin.writableEnded).toBe(false); + treeCleanup.resolve(true); + await flushMicrotasks(); + expect(child.stdin.writableEnded).toBe(true); + child.emit("exit", 0, null); + await closing; + }); + + test("ends stdin while POSIX process-tree cleanup is pending", async () => { + const child = new FakeMcpChild(); + const treeCleanup = Promise.withResolvers(); + const transport = new SupervisedStdioClientTransport( + { command: "mcp-server", cwd: "/workspace" }, + { + platform: "linux", + spawnProcess: () => child.asChildProcess(), + terminateProcessTree: async () => await treeCleanup.promise, + }, + ); + const starting = transport.start(); + child.emit("spawn"); + await starting; - expect(stdinEndedAtCleanup).toBe(false); + const closing = transport.close(); + expect(child.stdin.writableEnded).toBe(true); + treeCleanup.resolve(true); + child.emit("exit", 0, null); + await closing; }); test("fails cleanup when stdio process-tree termination is not verified", async () => { diff --git a/paseo-omp/tests/omp-rpc-process.test.ts b/paseo-omp/tests/omp-rpc-process.test.ts index a6334690..bd3d56c9 100644 --- a/paseo-omp/tests/omp-rpc-process.test.ts +++ b/paseo-omp/tests/omp-rpc-process.test.ts @@ -19,6 +19,8 @@ import { testOnWindows, } from "./helpers/omp-rpc-harness"; +const testOnPosix = process.platform === "win32" ? test.skip : test; + describe("OMP RPC transport", () => { testOnWindows("terminates a live Windows process tree with taskkill", async () => { const leader = spawn( @@ -50,6 +52,111 @@ process.stdout.write(String(descendant.pid) + "\\n", () => { ); }); + testOnWindows( + "keeps the Windows tree root alive until descendant cleanup completes", + async () => { + const script = ` + const { spawn } = require("node:child_process"); + const descendant = spawn( + process.execPath, + ["-e", "Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0)"], + { stdio: "ignore", windowsHide: true }, + ); + descendant.unref(); + process.stdout.write(JSON.stringify({ + type: "ready", + protocolVersion: 1, + supportedProtocolVersions: [1, 2], + maxFrameBytes: 1048576, + maxReassembledFrameBytes: 67108864, + }) + "\\n"); + let input = ""; + process.stdin.on("data", chunk => { + input += String(chunk); + while (true) { + const newline = input.indexOf("\\n"); + if (newline < 0) return; + const line = input.slice(0, newline); + input = input.slice(newline + 1); + if (!line) continue; + const command = JSON.parse(line); + if (command.type === "negotiate_protocol") { + process.stdout.write(JSON.stringify({ + type: "response", + id: command.id, + command: "negotiate_protocol", + success: true, + data: { protocolVersion: 2 }, + }) + "\\n"); + continue; + } + if (command.type !== "get_state") continue; + process.stdout.write(JSON.stringify({ + type: "notice", + level: "info", + message: String(descendant.pid), + }) + "\\n"); + process.stdout.write(JSON.stringify({ + type: "response", + id: command.id, + success: true, + data: { + model: null, + isStreaming: false, + isCompacting: false, + sessionId: "windows-tree", + }, + }) + "\\n"); + } + }); + process.stdin.on("end", () => process.exit(0)); + `; + let leader: ChildProcessWithoutNullStreams | undefined; + let descendantPid: number | undefined; + const runtime = new OmpRpcRuntime({ + spawnProcess(request) { + leader = spawn(process.execPath, ["-e", script], { + cwd: request.cwd, + env: { ...process.env, ...request.env }, + detached: request.detached, + stdio: ["pipe", "pipe", "pipe"], + windowsHide: true, + }); + return leader; + }, + async terminateProcessTree(pid) { + const root = leader; + if (!root) return false; + // Model a taskkill launch that loses the tree root when EOF is sent concurrently. + await Promise.race([once(root, "exit"), sleep(250)]); + return await terminateSpawnedProcessTree(pid, "win32"); + }, + environment: TEST_RUNTIME_ENV, + }); + + try { + const session = await runtime.startSession({ cwd: process.cwd(), mode: "full" }); + const descendantPidEvent = nextEvent((listener) => session.onEvent(listener)); + await session.getState(); + const notice = await descendantPidEvent; + if (notice.type !== "notice") throw new Error("Expected descendant PID notice"); + descendantPid = Number(notice.message); + if (!Number.isSafeInteger(descendantPid) || descendantPid < 1) { + throw new Error("Windows process-tree fixture did not report a valid descendant PID"); + } + const pid = descendantPid; + expect(() => process.kill(pid, 0)).not.toThrow(); + + await session.close(); + + expect(() => process.kill(pid, 0)).toThrow(expect.objectContaining({ code: "ESRCH" })); + } finally { + if (leader?.pid) await terminateSpawnedProcessTree(leader.pid, "win32"); + if (descendantPid) await terminateSpawnedProcessTree(descendantPid, "win32"); + } + }, + ); + test("terminates a surviving POSIX process group after its leader exited", async () => { const signals: Array = []; let descendantsAlive = true; @@ -148,8 +255,9 @@ process.stdout.write(String(descendant.pid) + "\\n", () => { expect(observed.filter((event) => event.type === "process_exit")).toHaveLength(1); }); - test("starts process-tree cleanup before stdin shutdown can release the Windows tree root", async () => { + testOnWindows("keeps stdin open until Windows process-tree cleanup settles", async () => { const child = new FakeRpcChild(); + const cleanup = Promise.withResolvers(); observeCommands(child, (command) => { if (command.type === "negotiate_protocol") { child.write({ @@ -160,22 +268,48 @@ process.stdout.write(String(descendant.pid) + "\\n", () => { }); } }); - let stdinEndedAtCleanup: boolean | undefined; const runtime = new OmpRpcRuntime({ spawnProcess: () => child.asChildProcess(), - terminateProcessTree: () => { - stdinEndedAtCleanup = child.stdin.writableEnded; - return Promise.resolve(true); - }, + terminateProcessTree: async () => await cleanup.promise, environment: TEST_RUNTIME_ENV, }); const opening = runtime.startSession({ cwd: "/repo", mode: "full" }); child.write(READY_FRAME); const session = await opening; - await session.close(); + const closing = session.close(); + expect(child.stdin.writableEnded).toBe(false); + cleanup.resolve(true); + await closing; + expect(child.stdin.writableEnded).toBe(true); + }); + + testOnPosix("ends stdin while POSIX process-tree cleanup is pending", async () => { + const child = new FakeRpcChild(); + const cleanup = Promise.withResolvers(); + observeCommands(child, (command) => { + if (command.type === "negotiate_protocol") { + child.write({ + type: "response", + id: command.id, + success: true, + data: { protocolVersion: 2 }, + }); + } + }); + const runtime = new OmpRpcRuntime({ + spawnProcess: () => child.asChildProcess(), + terminateProcessTree: async () => await cleanup.promise, + environment: TEST_RUNTIME_ENV, + }); + const opening = runtime.startSession({ cwd: "/repo", mode: "full" }); + child.write(READY_FRAME); + const session = await opening; - expect(stdinEndedAtCleanup).toBe(false); + const closing = session.close(); + expect(child.stdin.writableEnded).toBe(true); + cleanup.resolve(true); + await closing; }); test("surfaces unverified process-tree cleanup", async () => {