fix(core): clean up mcp stdio descendants (#34525)

packages/core/src/mcp/client.ts

@@ -1,6 +1,7 @@
 export * as MCPClient from "./client"
 
 import path from "node:path"
+import { execFile } from "node:child_process"
 import { pathToFileURL } from "node:url"
 import { Client, type ClientOptions } from "@modelcontextprotocol/sdk/client/index.js"
 import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
@@ -20,6 +21,8 @@ import { InstallationVersion } from "../installation/version"
 const DEFAULT_STARTUP_TIMEOUT = 30_000
 const DEFAULT_REQUEST_TIMEOUT = 30_000
 
+type Transport = StdioClientTransport | StreamableHTTPClientTransport
+
 // Some servers advertise tool outputSchemas the SDK's strict validator can't resolve; this drops
 // only that field so a single bad schema doesn't blank out the whole tool list.
 const TolerantListToolsResult = ListToolsResultSchema.extend({
@@ -73,7 +76,7 @@ export const connect = Effect.fnUntraced(function* (
   config: typeof ConfigMCP.Server.Type,
   directory: string,
 ) {
-  const transport = yield* Effect.gen(function* () {
+  const transport: Transport = yield* Effect.gen(function* () {
     if (config.type === "local") {
       const [command, ...args] = config.command
       return new StdioClientTransport({
@@ -111,7 +114,12 @@ export const connect = Effect.fnUntraced(function* (
     catch: (error) => error,
   }).pipe(Effect.exit)
   if (Exit.isSuccess(exit)) {
-    yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore))
+    yield* Effect.addFinalizer(() =>
+      cleanupStdioDescendants(transport).pipe(
+        Effect.andThen(Effect.promise(() => client.close())),
+        Effect.ignore,
+      ),
+    )
     const requestTimeout = config.timeout?.request ?? DEFAULT_REQUEST_TIMEOUT
     return {
       instructions: client.getInstructions()?.trim() || undefined,
@@ -185,13 +193,62 @@ export const connect = Effect.fnUntraced(function* (
     } satisfies Connection
   }
 
-  yield* Effect.promise(() => transport.close()).pipe(Effect.ignore)
+  yield* cleanupStdioDescendants(transport).pipe(
+    Effect.andThen(Effect.promise(() => transport.close())),
+    Effect.ignore,
+  )
   const error = Cause.squash(exit.cause)
   if (error instanceof UnauthorizedError || (error instanceof Error && error.message.includes("OAuth")))
     return yield* new NeedsAuthError({ server })
   return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) })
 })
 
+// SDK close stops the MCP process, but not child processes it spawned.
+const cleanupStdioDescendants = (transport: Transport) =>
+  Effect.gen(function* () {
+    if (!(transport instanceof StdioClientTransport)) return
+    const pid = transport.pid
+    if (typeof pid !== "number") return
+    yield* Effect.forEach(
+      yield* descendantPids(pid),
+      (pid) =>
+        Effect.try({
+          try: () => process.kill(pid, "SIGTERM"),
+          catch: () => undefined,
+        }).pipe(Effect.ignore),
+      { discard: true },
+    )
+  })
+
+const descendantPids = Effect.fnUntraced(function* (root: number) {
+  if (process.platform === "win32") return []
+  const result: number[] = []
+  const queue = [root]
+  for (let index = 0; index < queue.length; index++) {
+    const parent = queue[index]
+    if (parent === undefined) return result
+    const children = (yield* childPids(parent)).filter((pid) => !result.includes(pid))
+    result.push(...children)
+    queue.push(...children)
+  }
+  return result
+})
+
+const childPids = (pid: number) =>
+  Effect.promise(
+    () =>
+      new Promise<number[]>((resolve) => {
+        execFile("pgrep", ["-P", String(pid)], { encoding: "utf8" }, (_error, stdout) => {
+          resolve(
+            stdout
+              .split("\n")
+              .map((line) => Number.parseInt(line, 10))
+              .filter((pid) => Number.isInteger(pid)),
+          )
+        })
+      }),
+  )
+
 async function paginate<R extends { nextCursor?: string }, T>(
   list: (cursor: string | undefined) => Promise<R>,
   items: (result: R) => T[],