diff --git a/packages/opencode/src/mcp/index.ts b/packages/opencode/src/mcp/index.ts index 05f12fa2ee45..175be51276be 100644 --- a/packages/opencode/src/mcp/index.ts +++ b/packages/opencode/src/mcp/index.ts @@ -227,7 +227,7 @@ const layer = Layer.effect( }, catch: (e) => (e instanceof Error ? e : new Error(String(e))), }), - (t, exit) => (Exit.isFailure(exit) ? Effect.tryPromise(() => t.close()).pipe(Effect.ignore) : Effect.void), + (t, exit) => (Exit.isFailure(exit) ? stop(t) : Effect.void), ) }) @@ -398,11 +398,7 @@ const layer = Layer.effect( defs: listed, instructions: mcpClient.getInstructions()?.trim(), } satisfies CreateResult - }).pipe( - Effect.catchCause((cause) => - Effect.tryPromise(() => mcpClient.close()).pipe(Effect.ignore, Effect.andThen(Effect.failCause(cause))), - ), - ) + }).pipe(Effect.catchCause((cause) => stop(mcpClient).pipe(Effect.andThen(Effect.failCause(cause))))) }, Effect.map((result): CreateResult => result), Effect.catchCause((cause) => { @@ -439,6 +435,59 @@ const layer = Layer.effect( Effect.catch(() => Effect.succeed([] as number[])), ) + function running(pid: number) { + try { + process.kill(pid, 0) + return true + } catch (error) { + return !(error instanceof Error && "code" in error && error.code === "ESRCH") + } + } + + const waitForExit = Effect.fnUntraced(function* (pids: readonly number[], timeout: number) { + const deadline = Date.now() + timeout + let active = pids.filter(running) + while (active.length > 0 && Date.now() < deadline) { + yield* Effect.sleep("25 millis") + active = active.filter(running) + } + return active + }) + + function signal(pids: readonly number[], value: NodeJS.Signals) { + for (const pid of pids.toReversed()) { + try { + process.kill(pid, value) + } catch {} + } + } + + const stop = Effect.fnUntraced(function* ( + target: MCPClient | StdioClientTransport | TransportWithAuth | undefined, + ) { + if (!target) return + const transport = target instanceof Client ? target.transport : target + const root = transport instanceof StdioClientTransport ? transport.pid : null + if (typeof root !== "number") { + yield* Effect.tryPromise(() => target.close()).pipe(Effect.ignore) + return + } + const children = yield* descendants(root) + + if (children.length > 0) { + signal(children, "SIGTERM") + signal(yield* waitForExit(children, 500), "SIGKILL") + } + + yield* Effect.tryPromise(() => target.close()).pipe(Effect.ignore) + + const known = Array.from(new Set([root, ...children, ...(yield* descendants(root))])) + const remaining = yield* waitForExit(known, 250) + signal(remaining, "SIGKILL") + const survivors = yield* waitForExit(remaining, 1_000) + if (survivors.length > 0) yield* Effect.logWarning("MCP processes survived cleanup", { pids: survivors }) + }) + function watch(s: State, name: string, client: MCPClient, bridge: EffectBridge.Shape, timeout?: number) { client.onclose = () => { if (s.clients[name] !== client) return @@ -534,23 +583,7 @@ const layer = Layer.effect( s.clients = {} s.defs = {} s.instructions = {} - yield* Effect.forEach( - clients, - (client) => - Effect.gen(function* () { - const pid = client.transport instanceof StdioClientTransport ? client.transport.pid : null - if (typeof pid === "number") { - const pids = yield* descendants(pid) - for (const dpid of pids) { - try { - process.kill(dpid, "SIGTERM") - } catch {} - } - } - yield* Effect.tryPromise(() => client.close()).pipe(Effect.ignore) - }), - { concurrency: "unbounded" }, - ) + yield* Effect.forEach(clients, stop, { concurrency: "unbounded" }) pendingOAuthTransports.clear() }), ) @@ -565,7 +598,7 @@ const layer = Layer.effect( delete s.defs[name] delete s.instructions[name] if (!client) return Effect.void - return Effect.tryPromise(() => client.close()).pipe(Effect.ignore) + return stop(client) } const storeClient = Effect.fnUntraced(function* ( @@ -584,7 +617,7 @@ const layer = Layer.effect( if (instructions) s.instructions[name] = instructions else delete s.instructions[name] watch(s, name, client, bridge, timeout) - if (previous) yield* Effect.tryPromise(() => previous.close()).pipe(Effect.ignore) + if (previous) yield* stop(previous) return s.status[name] }) @@ -876,9 +909,7 @@ const layer = Layer.effect( const result = yield* startAuth(mcpName) if (!result.authorizationUrl) { const client = "client" in result ? result.client : undefined - const mcpConfig = yield* requireMcpConfig(mcpName).pipe( - Effect.tapError(() => Effect.tryPromise(() => client?.close() ?? Promise.resolve()).pipe(Effect.ignore)), - ) + const mcpConfig = yield* requireMcpConfig(mcpName).pipe(Effect.tapError(() => stop(client))) const listed = client ? client.getServerCapabilities()?.tools @@ -886,7 +917,7 @@ const layer = Layer.effect( : [] : undefined if (!client || !listed) { - yield* Effect.tryPromise(() => client?.close() ?? Promise.resolve()).pipe(Effect.ignore) + yield* stop(client) return { status: "failed", error: "Failed to get tools" } satisfies Status } diff --git a/packages/opencode/test/fixture/mcp-lifecycle-child.mjs b/packages/opencode/test/fixture/mcp-lifecycle-child.mjs new file mode 100644 index 000000000000..4ce9345d174a --- /dev/null +++ b/packages/opencode/test/fixture/mcp-lifecycle-child.mjs @@ -0,0 +1,24 @@ +import { spawn } from "node:child_process" +import { writeFileSync } from "node:fs" +import { fileURLToPath } from "node:url" + +const directory = process.env.MCP_LIFECYCLE_DIR +if (!directory) throw new Error("MCP_LIFECYCLE_DIR is required") + +const name = process.argv.includes("--grandchild") ? "grandchild" : "child" +writeFileSync(`${directory}/${name}.pid`, String(process.pid)) + +process.on("SIGTERM", () => { + writeFileSync(`${directory}/${name}.term`, "received") + if (name === "child") process.exit(0) +}) + +if (name === "child") { + const grandchild = spawn(process.execPath, [fileURLToPath(import.meta.url), "--grandchild"], { + env: process.env, + stdio: "ignore", + }) + void grandchild +} + +setInterval(() => {}, 60_000) diff --git a/packages/opencode/test/fixture/mcp-lifecycle-stdio.ts b/packages/opencode/test/fixture/mcp-lifecycle-stdio.ts index b01ed921cfd4..7c608e082cc8 100644 --- a/packages/opencode/test/fixture/mcp-lifecycle-stdio.ts +++ b/packages/opencode/test/fixture/mcp-lifecycle-stdio.ts @@ -1,6 +1,28 @@ import { Server } from "@modelcontextprotocol/sdk/server/index.js" import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js" import { ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js" +import { spawn } from "node:child_process" + +const lifecycleDir = process.env.MCP_LIFECYCLE_DIR +const subprocesses: ReturnType[] = [] + +async function waitFor(name: string) { + if (!lifecycleDir) throw new Error("MCP_LIFECYCLE_DIR is required") + while (!(await Bun.file(`${lifecycleDir}/${name}`).exists())) await Bun.sleep(10) +} + +if (process.argv.includes("--tree")) { + if (!lifecycleDir) throw new Error("MCP_LIFECYCLE_DIR is required") + await Bun.write(`${lifecycleDir}/parent.pid`, String(process.pid)) + const node = Bun.which("node") + if (!node) throw new Error("node is required for the MCP lifecycle fixture") + const subprocess = spawn(node, [`${import.meta.dir}/mcp-lifecycle-child.mjs`, "--child"], { + env: process.env, + stdio: ["ignore", "ignore", "inherit"], + }) + subprocesses.push(subprocess) + await waitFor("grandchild.pid") +} if (process.argv.includes("--hang")) { const pidFile = process.env.MCP_LIFECYCLE_PID_FILE @@ -11,8 +33,9 @@ if (process.argv.includes("--hang")) { const server = new Server({ name: "mcp-lifecycle-stdio", version: "1.0.0" }, { capabilities: { tools: {} } }) -server.setRequestHandler(ListToolsRequestSchema, () => - Promise.resolve({ +server.setRequestHandler(ListToolsRequestSchema, () => { + if (process.argv.includes("--list-error")) throw new Error("list tools failed") + return Promise.resolve({ tools: [ { name: "current_directory", @@ -20,7 +43,7 @@ server.setRequestHandler(ListToolsRequestSchema, () => inputSchema: { type: "object", properties: {} }, }, ], - }), -) + }) +}) await server.connect(new StdioServerTransport()) diff --git a/packages/opencode/test/mcp/lifecycle.test.ts b/packages/opencode/test/mcp/lifecycle.test.ts index 80c8fd22f886..e6a609af022a 100644 --- a/packages/opencode/test/mcp/lifecycle.test.ts +++ b/packages/opencode/test/mcp/lifecycle.test.ts @@ -20,6 +20,7 @@ import { MCP } from "../../src/mcp/index" import { McpOAuthCallback } from "../../src/mcp/oauth-callback" import { TestInstance } from "../fixture/fixture" import { pollWithTimeout, testEffect } from "../lib/effect" +import { InstanceStore } from "@/project/instance-store" const it = testEffect(LayerNode.compile(MCP.node)) const stdioFixture = path.join(import.meta.dir, "../fixture/mcp-lifecycle-stdio.ts") @@ -182,13 +183,55 @@ function statusName(status: Record | MCPNS.Status, server: const remote = (url: string, timeout?: number) => ({ type: "remote" as const, url, oauth: false as const, timeout }) +const treeFiles = ["parent", "child", "grandchild"] as const + +function treeCleanup(directory: string) { + return Effect.addFinalizer(() => + Effect.promise(async () => { + for (const name of treeFiles) { + const file = Bun.file(path.join(directory, `${name}.pid`)) + if (!(await file.exists())) continue + try { + process.kill(Number(await file.text()), "SIGKILL") + } catch {} + } + }), + ) +} + +function readTree(directory: string) { + return pollWithTimeout( + Effect.promise(async () => { + const files = treeFiles.map((name) => Bun.file(path.join(directory, `${name}.pid`))) + if (!(await Promise.all(files.map((file) => file.exists()))).every(Boolean)) return + return Promise.all(files.map(async (file) => Number(await file.text()))) + }), + "stdio fixture did not publish its process tree", + ) +} + +function processRunning(pid: number) { + try { + process.kill(pid, 0) + return true + } catch { + return false + } +} + +function waitForTreeExit(pids: readonly number[]) { + return pollWithTimeout( + Effect.sync(() => (pids.every((pid) => !processRunning(pid)) ? true : undefined)), + "stdio fixture process tree was not terminated", + ) +} + it.instance("advertises and lists the instance directory as its root", () => Effect.gen(function* () { const server = yield* lifecycleServer({ requestRoots: true }) const mcp = yield* MCP.Service const test = yield* TestInstance yield* mcp.add("roots", remote(server.url)) - const roots = yield* pollWithTimeout( Effect.sync(() => server.state.roots), "server did not receive roots", @@ -528,6 +571,99 @@ it.instance("local stdio timeout terminates the real server process", () => }), ) +if (process.platform !== "win32") { + it.instance( + "disconnect terminates a real stdio process tree", + () => + Effect.gen(function* () { + const test = yield* TestInstance + yield* treeCleanup(test.directory) + const mcp = yield* MCP.Service + const result = yield* mcp.add("tree-disconnect", { + type: "local", + command: [process.execPath, stdioFixture, "--tree"], + environment: { MCP_LIFECYCLE_DIR: test.directory }, + }) + expect(result.status).toMatchObject({ "tree-disconnect": { status: "connected" } }) + const pids = yield* readTree(test.directory) + expect(pids.every(processRunning)).toBe(true) + + yield* mcp.disconnect("tree-disconnect") + + yield* waitForTreeExit(pids) + expect(yield* Effect.promise(() => Bun.file(path.join(test.directory, "grandchild.term")).exists())).toBe(true) + }), + 20_000, + ) + + it.instance( + "replacement terminates only the old stdio process tree", + () => + Effect.gen(function* () { + const test = yield* TestInstance + yield* treeCleanup(test.directory) + const mcp = yield* MCP.Service + yield* mcp.add("tree-replace", { + type: "local", + command: [process.execPath, stdioFixture, "--tree"], + environment: { MCP_LIFECYCLE_DIR: test.directory }, + }) + const pids = yield* readTree(test.directory) + + const result = yield* mcp.add("tree-replace", { + type: "local", + command: [process.execPath, stdioFixture], + }) + + expect(statusName(result.status, "tree-replace")).toBe("connected") + yield* waitForTreeExit(pids) + expect(Object.keys(yield* mcp.tools())).toEqual(["tree-replace_current_directory"]) + }), + 20_000, + ) + + it.instance( + "tool discovery failure terminates a real stdio process tree", + () => + Effect.gen(function* () { + const test = yield* TestInstance + yield* treeCleanup(test.directory) + const mcp = yield* MCP.Service + const result = yield* mcp.add("tree-list-failure", { + type: "local", + command: [process.execPath, stdioFixture, "--tree", "--list-error"], + environment: { MCP_LIFECYCLE_DIR: test.directory }, + }) + const pids = yield* readTree(test.directory) + + expect(statusName(result.status, "tree-list-failure")).toBe("failed") + yield* waitForTreeExit(pids) + }), + 20_000, + ) + + it.instance( + "instance disposal waits for real stdio process tree cleanup", + () => + Effect.gen(function* () { + const test = yield* TestInstance + yield* treeCleanup(test.directory) + const mcp = yield* MCP.Service + yield* mcp.add("tree-dispose", { + type: "local", + command: [process.execPath, stdioFixture, "--tree"], + environment: { MCP_LIFECYCLE_DIR: test.directory }, + }) + const pids = yield* readTree(test.directory) + + yield* InstanceStore.Service.use((store) => store.disposeDirectory(test.directory)) + + yield* waitForTreeExit(pids) + }), + 20_000, + ) +} + it.instance("remote timeout aborts both real HTTP transport attempts", () => Effect.gen(function* () { const server = yield* hangingLifecycleServer()