Unified · Split
| @@ -0,0 +1,7 @@ | |||
| 1 | --- | ||
| 2 | "@executor-js/plugin-mcp": patch | ||
| 3 | --- | ||
| 4 | |||
| 5 | **Interrupted stdio dials no longer strand the spawned child process** | ||
| 6 | |||
| 7 | Cancelling an in-flight health check or tool discovery (a UI refresh aborting the request, or the 15s discovery timeout) abandoned the MCP connect handshake without closing the transport, leaving the spawned stdio child running indefinitely: for `docker run -i --rm` integrations, one stranded container per interrupted dial. The connect handshake now aborts on interruption (the SDK closes the transport, ending stdin and escalating to SIGTERM/SIGKILL), and tool discovery closes the connection even when the interrupt lands between the handshake completing and discovery starting. | ||
| @@ -257,7 +257,12 @@ const connectClient = (input: { · 256 unmodified lines | |||
| 257 | const transportInstance = input.createTransport(); | 257 | const transportInstance = input.createTransport(); |
| 258 | 258 | ||
| 259 | yield* Effect.tryPromise({ | 259 | yield* Effect.tryPromise({ |
| 260 | try: () => client.connect(transportInstance), | 260 | // Interruption (an HTTP 499 cancelling a health check, the discovery |
| 261 | // timeout) aborts this signal; the SDK then fails the in-flight | ||
| 262 | // handshake and closes the transport. Without it the abandoned connect | ||
| 263 | // kept the spawned stdio child alive forever; `docker run -i --rm` | ||
| 264 | // integrations stranded a container per interrupted dial (#1631). | ||
| 265 | try: (signal) => client.connect(transportInstance, { signal }), | ||
| 261 | catch: (cause) => | 266 | catch: (cause) => |
| 262 | connectionFailure(input.transport, `Failed connecting via ${input.transport}`, cause), | 267 | connectionFailure(input.transport, `Failed connecting via ${input.transport}`, cause), |
| 263 | }).pipe( | 268 | }).pipe( |
| @@ -91,34 +91,45 @@ const listAllTools = ( · 90 unmodified lines | |||
| 91 | * forever. On timeout, any connection that DID get established is closed | 91 | * forever. On timeout, any connection that DID get established is closed |
| 92 | * before the timeout error is raised (`Effect.onExit` still fires for an | 92 | * before the timeout error is raised (`Effect.onExit` still fires for an |
| 93 | * interrupted fiber). | 93 | * interrupted fiber). |
| 94 | * | ||
| 95 | * Interruption-safe: the connect phase cleans up after itself (the connector | ||
| 96 | * aborts the handshake and closes the transport, killing any spawned stdio | ||
| 97 | * child; #1631), and the mask below removes the window between the | ||
| 98 | * connector succeeding and `onExit` attaching, where an interrupt would | ||
| 99 | * leak the connection. The connector and listTools stay `restore`d so a 499 | ||
| 100 | * or the timeout above can still cancel them promptly. | ||
| 94 | */ | 101 | */ |
| 95 | export const discoverTools = ( | 102 | export const discoverTools = ( |
| 96 | connector: McpConnector, | 103 | connector: McpConnector, |
| 97 | timeoutMs: number = Duration.toMillis(DEFAULT_DISCOVER_TIMEOUT), | 104 | timeoutMs: number = Duration.toMillis(DEFAULT_DISCOVER_TIMEOUT), |
| 98 | ): Effect.Effect<McpToolManifest, McpToolDiscoveryError> => | 105 | ): Effect.Effect<McpToolManifest, McpToolDiscoveryError> => |
| 99 | Effect.gen(function* () { | 106 | Effect.uninterruptibleMask((restore) => |
| 100 | // Acquire connection | 107 | Effect.gen(function* () { |
| 101 | const connection = yield* connector.pipe( | 108 | // Acquire connection |
| 102 | Effect.mapError((failure) => { | 109 | const connection = yield* restore( |
| 103 | // Preserve the handshake HTTP status (401/403 = auth wall) so the | 110 | connector.pipe( |
| 104 | // liveness health check can classify structurally. | 111 | Effect.mapError((failure) => { |
| 105 | const httpStatus = Predicate.isTagged(failure, "McpConnectionError") | 112 | // Preserve the handshake HTTP status (401/403 = auth wall) so the |
| 106 | ? failure.httpStatus | 113 | // liveness health check can classify structurally. |
| 107 | : undefined; | 114 | const httpStatus = Predicate.isTagged(failure, "McpConnectionError") |
| 108 | return new McpToolDiscoveryError({ | 115 | ? failure.httpStatus |
| 109 | stage: "connect", | 116 | : undefined; |
| 110 | message: `Failed connecting to MCP server: ${failure.message}`, | 117 | return new McpToolDiscoveryError({ |
| 111 | ...(httpStatus !== undefined ? { httpStatus } : {}), | 118 | stage: "connect", |
| 112 | }); | 119 | message: `Failed connecting to MCP server: ${failure.message}`, |
| 113 | }), | 120 | ...(httpStatus !== undefined ? { httpStatus } : {}), |
| 114 | ); | 121 | }); |
| 122 | }), | ||
| 123 | ), | ||
| 124 | ); | ||
| 115 | 125 | ||
| 116 | const manifest = yield* listAllTools(connection).pipe( | 126 | const manifest = yield* restore(listAllTools(connection)).pipe( |
| 117 | Effect.onExit(() => closeConnection(connection)), | 127 | Effect.onExit(() => closeConnection(connection)), |
| 118 | ); | 128 | ); |
| 119 | 129 | ||
| 120 | return manifest; | 130 | return manifest; |
| 121 | }).pipe( | 131 | }), |
| 132 | ).pipe( | ||
| 122 | Effect.timeoutOrElse({ | 133 | Effect.timeoutOrElse({ |
| 123 | duration: Duration.millis(timeoutMs), | 134 | duration: Duration.millis(timeoutMs), |
| 124 | orElse: () => | 135 | orElse: () => |
| @@ -0,0 +1,112 @@ | |||
| 1 | // Regression coverage for #1631: interrupting a fiber mid-dial (an HTTP 499 | ||
| 2 | // cancelling a health check on app refresh, or the discovery timeout) must | ||
| 3 | // tear down the stdio child the transport spawned. Before the fix the | ||
| 4 | // abandoned `client.connect` promise kept the child alive forever; every | ||
| 5 | // interrupted health check stranded one `docker run -i --rm` container. | ||
| 6 | // | ||
| 7 | // `it.live`: these tests measure real child-process lifetime, so they need | ||
| 8 | // the wall clock; under the TestClock the fixture's delayed initialize | ||
| 9 | // reply and the discovery timeout would never fire. | ||
| 10 | |||
| 11 | import { describe, expect, it } from "@effect/vitest"; | ||
| 12 | import { Duration, Effect, Fiber } from "effect"; | ||
| 13 | import { existsSync, mkdtempSync, readFileSync } from "node:fs"; | ||
| 14 | import { tmpdir } from "node:os"; | ||
| 15 | import { join } from "node:path"; | ||
| 16 | import { fileURLToPath } from "node:url"; | ||
| 17 | |||
| 18 | import { createMcpConnector } from "./connection"; | ||
| 19 | import { discoverTools } from "./discover"; | ||
| 20 | |||
| 21 | const fixture = fileURLToPath(new URL("./stdio-interrupt-test-server.ts", import.meta.url)); | ||
| 22 | |||
| 23 | const isAlive = (pid: number): boolean => { | ||
| 24 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: process.kill(pid, 0) reports "process gone" only by throwing ESRCH | ||
| 25 | try { | ||
| 26 | process.kill(pid, 0); | ||
| 27 | return true; | ||
| 28 | } catch { | ||
| 29 | return false; | ||
| 30 | } | ||
| 31 | }; | ||
| 32 | |||
| 33 | const killQuietly = (pid: number): void => { | ||
| 34 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: kill throws ESRCH when the child already exited, which is the desired state | ||
| 35 | try { | ||
| 36 | process.kill(pid, "SIGKILL"); | ||
| 37 | } catch { | ||
| 38 | // already gone | ||
| 39 | } | ||
| 40 | }; | ||
| 41 | |||
| 42 | const waitUntil = (predicate: () => boolean, timeoutMs: number) => | ||
| 43 | Effect.gen(function* () { | ||
| 44 | const deadline = Date.now() + timeoutMs; | ||
| 45 | while (!predicate()) { | ||
| 46 | if (Date.now() > deadline) return false; | ||
| 47 | yield* Effect.sleep(Duration.millis(50)); | ||
| 48 | } | ||
| 49 | return true; | ||
| 50 | }); | ||
| 51 | |||
| 52 | const makeFixture = (mode: "fast" | "slow" | "never") => { | ||
| 53 | const pidFile = join(mkdtempSync(join(tmpdir(), "mcp-stdio-interrupt-")), "pid"); | ||
| 54 | const connector = createMcpConnector({ | ||
| 55 | transport: "stdio", | ||
| 56 | command: "bun", | ||
| 57 | args: ["run", fixture, pidFile, mode], | ||
| 58 | }); | ||
| 59 | const spawned = waitUntil(() => existsSync(pidFile), 10_000); | ||
| 60 | const readPid = () => Number(readFileSync(pidFile, "utf8")); | ||
| 61 | return { connector, spawned, readPid }; | ||
| 62 | }; | ||
| 63 | |||
| 64 | // The transport's teardown ends stdin first and escalates to SIGTERM only | ||
| 65 | // after 2s, so a cleaned-up child can legitimately take a moment to exit. | ||
| 66 | const exitsAfterCleanup = (pid: number) => | ||
| 67 | Effect.gen(function* () { | ||
| 68 | const exited = yield* waitUntil(() => !isAlive(pid), 5_000); | ||
| 69 | killQuietly(pid); | ||
| 70 | return exited; | ||
| 71 | }); | ||
| 72 | |||
| 73 | describe("stdio child cleanup on interruption (#1631)", () => { | ||
| 74 | it.live("uninterrupted discovery closes the child", () => | ||
| 75 | Effect.gen(function* () { | ||
| 76 | const { connector, spawned, readPid } = makeFixture("fast"); | ||
| 77 | const manifest = yield* discoverTools(connector); | ||
| 78 | expect(manifest.tools).toEqual([]); | ||
| 79 | expect(yield* spawned).toBe(true); | ||
| 80 | expect(yield* exitsAfterCleanup(readPid())).toBe(true); | ||
| 81 | }), | ||
| 82 | ); | ||
| 83 | |||
| 84 | it.live("interrupting mid-handshake kills the child", () => | ||
| 85 | Effect.gen(function* () { | ||
| 86 | const { connector, spawned, readPid } = makeFixture("slow"); | ||
| 87 | const fiber = yield* discoverTools(connector).pipe(Effect.forkDetach); | ||
| 88 | |||
| 89 | expect(yield* spawned).toBe(true); | ||
| 90 | const pid = readPid(); | ||
| 91 | expect(isAlive(pid)).toBe(true); | ||
| 92 | |||
| 93 | // The initialize reply arrives at t+3s, so the handshake is still in | ||
| 94 | // flight; cancel the fiber the way the HTTP layer does on a 499. | ||
| 95 | yield* Effect.sleep(Duration.millis(200)); | ||
| 96 | yield* Fiber.interrupt(fiber); | ||
| 97 | |||
| 98 | expect(yield* exitsAfterCleanup(pid)).toBe(true); | ||
| 99 | }), | ||
| 100 | ); | ||
| 101 | |||
| 102 | it.live("the discovery timeout kills the child", () => | ||
| 103 | Effect.gen(function* () { | ||
| 104 | const { connector, spawned, readPid } = makeFixture("never"); | ||
| 105 | const failure = yield* discoverTools(connector, 1_000).pipe(Effect.flip); | ||
| 106 | expect(failure.message).toContain("timed out"); | ||
| 107 | |||
| 108 | expect(yield* spawned).toBe(true); | ||
| 109 | expect(yield* exitsAfterCleanup(readPid())).toBe(true); | ||
| 110 | }), | ||
| 111 | ); | ||
| 112 | }); | ||
| @@ -0,0 +1,65 @@ | |||
| 1 | // Fixture for stdio-interrupt-cleanup.test.ts. A minimal legacy-handshake MCP | ||
| 2 | // server that stands in for a `docker run -i --rm` stdio integration: it | ||
| 3 | // writes its PID to the file named by argv so the test can observe process | ||
| 4 | // lifetime, and it exits only when stdin closes or it is signalled (the same | ||
| 5 | // exit contract as the docker CLI). The mode argument controls the initialize | ||
| 6 | // reply: "fast" answers immediately, "slow" answers after 3s (keeps the | ||
| 7 | // handshake in flight so the test can interrupt mid-connect), "never" withholds | ||
| 8 | // it (a wedged server, for the discovery-timeout path). | ||
| 9 | |||
| 10 | import { writeFileSync } from "node:fs"; | ||
| 11 | |||
| 12 | const pidFile = process.argv[2]; | ||
| 13 | const mode = process.argv[3] ?? "fast"; | ||
| 14 | if (pidFile === undefined) { | ||
| 15 | process.stderr.write("usage: stdio-interrupt-test-server.ts <pid-file> [fast|slow|never]\n"); | ||
| 16 | process.exit(2); | ||
| 17 | } | ||
| 18 | writeFileSync(pidFile, String(process.pid)); | ||
| 19 | |||
| 20 | const respond = (message: object): void => { | ||
| 21 | process.stdout.write(`${JSON.stringify(message)}\n`); | ||
| 22 | }; | ||
| 23 | |||
| 24 | const handle = (line: string): void => { | ||
| 25 | if (!line.trim()) return; | ||
| 26 | let request: { id?: number; method?: string; params?: { protocolVersion?: string } }; | ||
| 27 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: standalone non-Effect fixture process; a malformed frame is silently dropped like a real server would | ||
| 28 | try { | ||
| 29 | // oxlint-disable-next-line executor/no-json-parse -- boundary: hand-rolled JSON-RPC framing is the fixture's entire purpose (it must control handshake timing below the SDK) | ||
| 30 | request = JSON.parse(line); | ||
| 31 | } catch { | ||
| 32 | return; | ||
| 33 | } | ||
| 34 | if (request.method === "initialize") { | ||
| 35 | const reply = () => | ||
| 36 | respond({ | ||
| 37 | jsonrpc: "2.0", | ||
| 38 | id: request.id, | ||
| 39 | result: { | ||
| 40 | protocolVersion: request.params?.protocolVersion, | ||
| 41 | capabilities: { tools: {} }, | ||
| 42 | serverInfo: { name: "stdio-interrupt-test-server", version: "0.0.0" }, | ||
| 43 | }, | ||
| 44 | }); | ||
| 45 | if (mode === "slow") setTimeout(reply, 3_000); | ||
| 46 | else if (mode !== "never") reply(); | ||
| 47 | } else if (request.method === "tools/list") { | ||
| 48 | respond({ jsonrpc: "2.0", id: request.id, result: { tools: [] } }); | ||
| 49 | } else if (request.id !== undefined) { | ||
| 50 | respond({ jsonrpc: "2.0", id: request.id, result: {} }); | ||
| 51 | } | ||
| 52 | }; | ||
| 53 | |||
| 54 | let buffer = ""; | ||
| 55 | process.stdin.setEncoding("utf8"); | ||
| 56 | process.stdin.on("data", (chunk: string) => { | ||
| 57 | buffer += chunk; | ||
| 58 | let newline = buffer.indexOf("\n"); | ||
| 59 | while (newline >= 0) { | ||
| 60 | handle(buffer.slice(0, newline)); | ||
| 61 | buffer = buffer.slice(newline + 1); | ||
| 62 | newline = buffer.indexOf("\n"); | ||
| 63 | } | ||
| 64 | }); | ||
| 65 | process.stdin.on("end", () => process.exit(0)); | ||