refactor(core): spawn stdio MCP servers through the location environment (#42076)

packages/core/src/mcp/client.ts

@@ -1,11 +1,10 @@
 export * as MCPClient from "./client.js"
 
 import path from "node:path"
-import { execFile } from "node:child_process"
 import { pathToFileURL } from "node:url"
 import { Client } from "@modelcontextprotocol/sdk/client/index.js"
-import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
 import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"
+import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"
 import { UnauthorizedError, type OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth.js"
 import {
   CallToolResultSchema,
@@ -30,13 +29,12 @@ import {
 } from "@modelcontextprotocol/sdk/types.js"
 import { Cause, Effect, Exit, Schema } from "effect"
 import { ConfigMCP } from "@opencode-ai/schema/config/mcp"
+import { MCPStdio } from "./stdio.js"
 
 const DEFAULT_STARTUP_TIMEOUT = 30_000
 const DEFAULT_CATALOG_TIMEOUT = 30_000
 const DEFAULT_EXECUTION_TIMEOUT = 12 * 60 * 60 * 1_000 // 12 hours
 
-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({
@@ -176,7 +174,12 @@ export interface Connection {
   readonly onResourcesChanged: (callback: () => void) => void
 }
 
-/** Connects an MCP server; closing the calling scope tears down the transport and any spawned process. */
+/**
+ * Connects an MCP server; closing the calling scope tears down the transport and any spawned process.
+ *
+ * A stdio server is spawned through the location's `Environment`, so it runs on the same execution
+ * plane as the location's shell commands rather than always on the host.
+ */
 export const connect = Effect.fnUntraced(function* (
   server: string,
   config: typeof ConfigMCP.Server.Type,
@@ -190,13 +193,12 @@ export const connect = Effect.fnUntraced(function* (
   const transport: Transport = yield* Effect.gen(function* () {
     if (config.type === "local") {
       const [command, ...args] = config.command
-      return new StdioClientTransport({
+      return yield* MCPStdio.make({
+        server,
         command,
         args,
         cwd: config.cwd ? path.resolve(directory, config.cwd) : directory,
-        stderr: "pipe",
-        env: {
-          ...(process.env as Record<string, string>),
+        environment: {
           ...(command === "opencode" ? { BUN_BE_BUN: "1" } : {}),
           ...config.environment,
         },
@@ -233,9 +235,9 @@ export const connect = Effect.fnUntraced(function* (
     catch: (error) => error,
   }).pipe(Effect.exit)
   if (Exit.isSuccess(exit)) {
-    yield* Effect.addFinalizer(() =>
-      cleanupStdioDescendants(transport).pipe(Effect.andThen(Effect.promise(() => client.close())), Effect.ignore),
-    )
+    // Closing the client closes the transport, which ends stdin and then kills through the spawner
+    // handle if the server does not exit cleanly. The process scope remains a final backstop.
+    yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore))
     const catalogTimeout = config.timeout?.catalog ?? DEFAULT_CATALOG_TIMEOUT
     const executionTimeout = config.timeout?.execution ?? DEFAULT_EXECUTION_TIMEOUT
     return {
@@ -434,58 +436,12 @@ export const connect = Effect.fnUntraced(function* (
     } satisfies Connection
   }
 
-  yield* cleanupStdioDescendants(transport).pipe(Effect.andThen(Effect.promise(() => transport.close())), Effect.ignore)
+  yield* Effect.promise(() => transport.close()).pipe(Effect.ignore)
   const error = Cause.squash(exit.cause)
   if (error instanceof UnauthorizedError) 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[],

packages/core/src/mcp/index.ts

@@ -12,6 +12,7 @@ import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import { Config } from "../config.js"
 import { Credential } from "../credential.js"
 import { Bus } from "../bus.js"
+import { Environment } from "../environment/index.js"
 import { Form } from "../form.js"
 import { Integration } from "../integration.js"
 import { KeyedMutex } from "../effect/keyed-mutex.js"
@@ -173,13 +174,13 @@ export const layer = (options?: Options) =>
     Effect.gen(function* () {
       const config = yield* Config.Service
       const location = yield* Location.Service
+      const environment = yield* Environment.Service
       const bus = yield* Bus.Service
       const forms = yield* Form.Service
       const integration = yield* Integration.Service
       const credentials = yield* Credential.Service
-      const root = yield* Scope.make()
+      const root = yield* Effect.scope
       const fork = yield* FiberSet.makeRuntime<never, void, never>()
-      yield* Effect.addFinalizer((exit) => Scope.close(root, exit))
 
       const loadConfig = (entries: readonly Entry[]) => {
         const documents = entries.filter((entry): entry is Document => entry.type === "document")
@@ -459,13 +460,8 @@ export const layer = (options?: Options) =>
         connection.onClose(() =>
           live(
             Effect.gen(function* () {
-              entry.client = undefined
-              entry.tools = undefined
-              entry.prompts = undefined
               entry.status = { status: "failed", error: "Connection closed" }
-              yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
-              yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
-              yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore)
+              yield* stopServer(name, entry)
               yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
             }),
           ),
@@ -520,6 +516,8 @@ export const layer = (options?: Options) =>
             options?.clientInfo,
           ).pipe(
             Effect.flatMap((connection) => connection.tools().pipe(Effect.map((tools) => ({ connection, tools })))),
+            // A stdio server is spawned on this location's execution plane, not the host's.
+            Effect.provideService(Environment.Service, environment),
             Scope.provide(scope),
             Effect.exit,
           )
@@ -828,7 +826,7 @@ export function configured(options?: Options) {
   return makeLocationNode({
     service: Service,
     layer: layer(options),
-    deps: [Config.node, Location.node, Bus.node, Form.node, Integration.node, Credential.node],
+    deps: [Config.node, Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node],
   })
 }
 

packages/core/src/mcp/stdio.ts

@@ -0,0 +1,181 @@
+export * as MCPStdio from "./stdio.js"
+
+import { ReadBuffer, serializeMessage } from "@modelcontextprotocol/sdk/shared/stdio.js"
+import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"
+import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js"
+import { Cause, Duration, Effect, Queue, Scope, Stream } from "effect"
+import { ChildProcess } from "effect/unstable/process"
+import type { ChildProcessHandle } from "effect/unstable/process/ChildProcessSpawner"
+import { Environment } from "../environment/index.js"
+
+/** Mirrors StdioClientTransport: wait this long for a graceful exit after stdin closes. */
+const CLOSE_GRACE = Duration.seconds(2)
+
+/** Mirrors StdioClientTransport: escalate SIGTERM to SIGKILL after this long. */
+const FORCE_KILL_AFTER = Duration.seconds(2)
+const OUTGOING_CAPACITY = 64
+const MAX_FRAME_BYTES = 16 * 1024 * 1024
+
+export interface Options {
+  /** Server name; only used to attribute logs. */
+  readonly server: string
+  readonly command: string
+  readonly args: ReadonlyArray<string>
+  readonly cwd: string
+  /**
+   * Environment declared by the server config, and nothing else.
+   *
+   * The host environment is merged in by the spawner via `extendEnv`, which keeps the merge on the
+   * side that actually runs the process: the local driver extends with the host's `process.env`
+   * (what the MCP SDK's transport did), while a workspace driver extends with the sandbox's own
+   * environment. Host variables therefore never cross the seam into a remote workspace.
+   */
+  readonly environment: Record<string, string>
+}
+
+/**
+ * MCP stdio transport that spawns its server through the location's `Environment` instead of the
+ * SDK's host-bound `StdioClientTransport`, so a workspace-backed location runs its MCP servers
+ * wherever the rest of its execution happens.
+ *
+ * The process is acquired in the calling scope: closing the scope kills it (the spawner kills the
+ * whole process group, so descendants go too) regardless of whether the transport was closed.
+ */
+export const make = Effect.fnUntraced(function* (options: Options) {
+  const environment = yield* Environment.Service
+  const scope = yield* Effect.scope
+  // Outgoing frames are queued rather than written to `handle.stdin` directly: the sink closes the
+  // stream it is run with, and stdin must stay open across the whole session.
+  const outgoing = yield* Queue.bounded<string, Cause.Done>(OUTGOING_CAPACITY)
+  const buffer = new ReadBuffer()
+  const state: { phase: "ready" | "starting" | "open" | "closed"; handle?: ChildProcessHandle } = { phase: "ready" }
+  let startup: Promise<void> | undefined
+  let closing: Promise<void> | undefined
+  let trailingBytes = 0
+
+  const stop = (handle: ChildProcessHandle) =>
+    Effect.gen(function* () {
+      const exit = yield* Effect.timeoutOption(handle.exitCode, CLOSE_GRACE)
+      if (exit._tag === "Some") return
+      const terminated = yield* Effect.timeoutOption(handle.kill({ killSignal: "SIGTERM" }), FORCE_KILL_AFTER)
+      if (terminated._tag === "None") yield* handle.kill({ killSignal: "SIGKILL" })
+    }).pipe(Effect.ignore)
+
+  const close = () =>
+    (closing ??= Effect.runPromise(
+      Effect.gen(function* () {
+        state.phase = "closed"
+        Queue.endUnsafe(outgoing)
+        if (startup) yield* Effect.promise(() => startup!.catch(() => undefined))
+        const handle = state.handle
+        if (!handle) return
+        state.handle = undefined
+        yield* stop(handle)
+      }).pipe(Effect.ensuring(Queue.shutdown(outgoing)), Effect.ensuring(Effect.sync(() => buffer.clear()))),
+    ))
+
+  const transport: Transport = {
+    start: () => {
+      if (state.phase !== "ready") return Promise.reject(new Error("Stdio transport already started"))
+      state.phase = "starting"
+      startup = Effect.runPromise(
+        Effect.gen(function* () {
+          const handle = yield* environment.spawner.spawn(
+            ChildProcess.make(options.command, [...options.args], {
+              cwd: options.cwd,
+              env: options.environment,
+              extendEnv: true,
+              stdin: { stream: Stream.encodeText(Stream.fromQueue(outgoing)), endOnDone: true },
+              stdout: "pipe",
+              stderr: "pipe",
+              forceKillAfter: FORCE_KILL_AFTER,
+            }),
+          )
+          state.handle = handle
+          if (state.phase === "closed") {
+            state.handle = undefined
+            return yield* stop(handle)
+          }
+          state.phase = "open"
+          yield* startOutput(handle)
+        }).pipe(Scope.provide(scope)),
+      )
+      return startup
+    },
+    send: (message: JSONRPCMessage) =>
+      state.phase !== "open"
+        ? Promise.reject(new Error("Not connected"))
+        : Effect.runPromise(
+            Queue.offer(outgoing, serializeMessage(message)).pipe(
+              Effect.flatMap((offered) => (offered ? Effect.void : Effect.fail(new Error("Not connected")))),
+            ),
+          ),
+    close,
+  }
+
+  const deliver = (chunk: Uint8Array) =>
+    Effect.gen(function* () {
+      for (const byte of chunk) {
+        trailingBytes = byte === 10 ? 0 : trailingBytes + 1
+        if (trailingBytes > MAX_FRAME_BYTES) return yield* Effect.fail(new Error("MCP stdio frame exceeded 16 MiB"))
+      }
+      buffer.append(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength))
+      while (true) {
+        // `undefined` means the frame failed to parse: the buffer has already advanced past it, so
+        // keep draining. `null` means the buffer holds no complete frame yet.
+        const message = yield* Effect.try({
+          try: () => buffer.readMessage(),
+          catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
+        }).pipe(
+          Effect.catch((error) =>
+            Effect.sync(() => {
+              transport.onerror?.(error)
+              return undefined
+            }),
+          ),
+        )
+        if (message === undefined) continue
+        if (message === null) return
+        transport.onmessage?.(message)
+      }
+    })
+
+  const startOutput = (handle: ChildProcessHandle) =>
+    Effect.gen(function* () {
+      yield* Effect.forkScoped(
+        Stream.runForEach(handle.stdout, deliver).pipe(
+          Effect.tapCause((cause) =>
+            Effect.sync(() => {
+              const error = Cause.squash(cause)
+              transport.onerror?.(error instanceof Error ? error : new Error(String(error)))
+            }),
+          ),
+          Effect.ignore,
+          // stdout ending means the server is gone; the SDK transport reports that the same way.
+          Effect.ensuring(
+            Effect.gen(function* () {
+              const unexpected = state.phase !== "closed"
+              if (unexpected) yield* Effect.promise(close)
+              transport.onclose?.()
+            }),
+          ),
+        ),
+      )
+
+      // StdioClientTransport pipes stderr into a stream nobody reads. Drain chunks into the debug
+      // log so chatty servers cannot stall and newline-free output is not buffered without bound.
+      yield* Effect.forkScoped(
+        handle.stderr.pipe(
+          Stream.decodeText(),
+          Stream.runForEach((output) =>
+            output.trim() === ""
+              ? Effect.void
+              : Effect.logDebug("mcp server stderr", { server: options.server, output }),
+          ),
+          Effect.ignore,
+        ),
+      )
+    })
+
+  return transport
+})

packages/core/test/fixture/environment.ts

@@ -1,7 +1,42 @@
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { Environment } from "@opencode-ai/core/environment/index"
 import { Location } from "@opencode-ai/core/location"
+import { CrossSpawnSpawner } from "@opencode-ai/util/cross-spawn-spawner"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
 import { Effect, Layer } from "effect"
+import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
+
+/**
+ * The host environment, without the workspace machinery: what a location with no `workspaceID`
+ * resolves to.
+ */
+export const hostEnvironmentLayer = Layer.effect(
+  Environment.Service,
+  Effect.gen(function* () {
+    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
+    const driver = Environment.makeLocalDriver(spawner)
+    return Environment.Service.of({ files: Environment.makeFiles(driver), spawner: driver.spawner })
+  }),
+).pipe(Layer.provide(LayerNode.compile(CrossSpawnSpawner.node)))
+
+/**
+ * The host environment with its spawner wrapped so a test can assert on every command that crosses
+ * the seam. Spawning still really happens, so the process under test behaves normally.
+ */
+export const recordingEnvironmentLayer = (spawns: Array<ChildProcess.Command>) =>
+  Layer.effect(
+    Environment.Service,
+    Effect.gen(function* () {
+      const environment = yield* Environment.Service
+      return Environment.Service.of({
+        ...environment,
+        spawner: ChildProcessSpawner.make((command) => {
+          spawns.push(command)
+          return environment.spawner.spawn(command)
+        }),
+      })
+    }),
+  ).pipe(Layer.provide(hostEnvironmentLayer))
 
 export type EnvironmentFilesTransform = (files: Environment.Files) => Partial<Environment.Files>
 

packages/core/test/mcp.test.ts

@@ -22,19 +22,24 @@ import { Bus } from "@opencode-ai/core/bus"
 import { ID, type Payload } from "@opencode-ai/schema/event"
 import { Form } from "@opencode-ai/core/form"
 import { Integration } from "@opencode-ai/core/integration"
+import { Environment } from "@opencode-ai/core/environment/index"
 import { Location } from "@opencode-ai/core/location"
 import { MCP } from "@opencode-ai/core/mcp/index"
 import { MCPClient } from "@opencode-ai/core/mcp/client"
+import { MCPStdio } from "@opencode-ai/core/mcp/stdio"
 import { Permission } from "@opencode-ai/core/permission"
 import { AbsolutePath } from "@opencode-ai/core/schema"
 import { Session } from "@opencode-ai/core/session"
 import { McpTool } from "@opencode-ai/core/tool/mcp"
 import { Tool } from "@opencode-ai/core/tool"
-import { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Schedule, Schema, Stream } from "effect"
+import { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Schedule, Schema, Sink, Stream } from "effect"
+import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
+import { ExitCode, makeHandle, ProcessId } from "effect/unstable/process/ChildProcessSpawner"
 import { Image } from "@opencode-ai/core/image"
 import { testEffect } from "./lib/effect"
 import { imagePassthrough } from "./lib/image"
 import { location } from "./fixture/location"
+import { hostEnvironmentLayer, recordingEnvironmentLayer } from "./fixture/environment"
 import { executeTool, toolDefinitions, toolIdentity, waitForCodeModeTool, waitForTool } from "./lib/tool"
 
 let assertion: Deferred.Deferred<Permission.AssertInput> | undefined
@@ -167,6 +172,7 @@ function resourceMcpLayer(
   overrides?: {
     entries?: Config.Interface["entries"]
     subscribe?: Bus.Interface["subscribe"]
+    environment?: Layer.Layer<Environment.Service>
   },
 ) {
   const directory = AbsolutePath.make(import.meta.dir)
@@ -229,11 +235,15 @@ function resourceMcpLayer(
           },
         }),
         Layer.mock(Credential.Service, {}),
+        overrides?.environment ?? hostEnvironmentLayer,
       ),
     ),
   )
 }
 
+const connect = (server: string, config: typeof ConfigMCP.Server.Type, directory: string) =>
+  MCPClient.connect(server, config, directory).pipe(Effect.provide(hostEnvironmentLayer))
+
 const mcp = Layer.mock(MCP.Service, {
   tools: () =>
     Effect.succeed([
@@ -407,7 +417,7 @@ test("retains output schemas across paginated MCP discovery", async () => {
   const tools = await Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "pagination",
           new ConfigMCP.Local({
             type: "local",
@@ -440,11 +450,139 @@ test("retains output schemas across paginated MCP discovery", async () => {
   ])
 })
 
+test("spawns local MCP servers through the location environment", async () => {
+  const spawns: Array<ChildProcess.Command> = []
+  const cwd = path.join(import.meta.dir, "fixture")
+  const config = new ConfigMCP.Local({
+    type: "local",
+    command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
+    cwd: "fixture",
+    environment: { MCP_LOCATION_TEST: "configured" },
+  })
+
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const connection = yield* MCPClient.connect("environment", config, import.meta.dir)
+        yield* connection.tools()
+      }),
+    ).pipe(Effect.provide(recordingEnvironmentLayer(spawns))),
+  )
+
+  expect(spawns).toHaveLength(1)
+  const command = spawns[0]
+  if (!command || !ChildProcess.isStandardCommand(command)) throw new Error("Expected a standard process command")
+  expect(command.command).toBe(process.execPath)
+  expect(command.options.cwd).toBe(cwd)
+  expect(command.options.extendEnv).toBe(true)
+  expect(command.options.env).toEqual({ MCP_LOCATION_TEST: "configured" })
+})
+
+test("rejects sends before the stdio transport is started", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const transport = yield* MCPStdio.make({
+          server: "not-started",
+          command: process.execPath,
+          args: [path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
+          cwd: import.meta.dir,
+          environment: {},
+        })
+        yield* Effect.tryPromise({
+          try: () => transport.send({ jsonrpc: "2.0", method: "notifications/initialized" }),
+          catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
+        }).pipe(
+          Effect.flip,
+          Effect.tap((error) => Effect.sync(() => expect(error.message).toBe("Not connected"))),
+        )
+      }).pipe(Effect.provide(hostEnvironmentLayer)),
+    ),
+  )
+})
+
+test("joins concurrent stdio transport closes", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const transport = yield* MCPStdio.make({
+          server: "concurrent-close",
+          command: "unused",
+          args: [],
+          cwd: import.meta.dir,
+          environment: {},
+        })
+        const first = transport.close()
+        expect(transport.close()).toBe(first)
+        yield* Effect.promise(() => first)
+      }).pipe(Effect.provide(hostEnvironmentLayer)),
+    ),
+  )
+})
+
+test("closes a stdio process that finishes spawning after close", async () => {
+  const spawning = Deferred.makeUnsafe<void>()
+  const release = Deferred.makeUnsafe<void>()
+  const exited = Deferred.makeUnsafe<ExitCode>()
+  const signals: Array<string> = []
+  const driver = Environment.makeMemoryDriver()
+  const environment = Layer.succeed(
+    Environment.Service,
+    Environment.Service.of({
+      files: Environment.makeFiles(driver),
+      spawner: ChildProcessSpawner.make(() =>
+        Effect.gen(function* () {
+          yield* Deferred.succeed(spawning, undefined)
+          yield* Deferred.await(release)
+          return makeHandle({
+            pid: ProcessId(1),
+            exitCode: Deferred.await(exited),
+            isRunning: Deferred.isDone(exited).pipe(Effect.map((done) => !done)),
+            kill: (options) =>
+              Effect.gen(function* () {
+                signals.push(options?.killSignal ?? "SIGTERM")
+                yield* Deferred.succeed(exited, ExitCode(143))
+              }),
+            stdin: Sink.drain,
+            stdout: Stream.never,
+            stderr: Stream.empty,
+            all: Stream.never,
+            getInputFd: () => Sink.drain,
+            getOutputFd: () => Stream.empty,
+            unref: Effect.succeed(Effect.void),
+          })
+        }),
+      ),
+    }),
+  )
+
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const transport = yield* MCPStdio.make({
+          server: "close-during-spawn",
+          command: "unused",
+          args: [],
+          cwd: import.meta.dir,
+          environment: {},
+        })
+        const start = transport.start()
+        yield* Deferred.await(spawning)
+        const close = transport.close()
+        yield* Deferred.succeed(release, undefined)
+        yield* Effect.promise(() => Promise.all([start, close]))
+      }).pipe(Effect.provide(environment)),
+    ),
+  )
+
+  expect(signals).toEqual(["SIGTERM"])
+})
+
 test("applies the configured MCP catalog timeout", async () => {
   const result = Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "catalog-timeout",
           new ConfigMCP.Local({
             type: "local",
@@ -466,7 +604,7 @@ test("applies the configured MCP execution timeout", async () => {
   const result = Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "execution-timeout",
           new ConfigMCP.Local({
             type: "local",
@@ -487,7 +625,7 @@ test("applies the configured MCP execution timeout to prompts", async () => {
   const result = Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "prompt-timeout",
           new ConfigMCP.Local({
             type: "local",
@@ -508,7 +646,7 @@ test("applies configured MCP timeouts to resource operations", async () => {
   const catalog = Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "resource-catalog-timeout",
           new ConfigMCP.Local({
             type: "local",
@@ -527,7 +665,7 @@ test("applies configured MCP timeouts to resource operations", async () => {
   const read = Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "resource-read-timeout",
           new ConfigMCP.Local({
             type: "local",
@@ -562,7 +700,7 @@ test("lists, reads, and reports MCP resource changes", async () => {
           },
           "templates-2": { items: [{ name: "Issue", uriTemplate: "issue://{id}", description: "Issue" }] },
         }
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "resources",
           new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }),
           import.meta.dir,
@@ -633,7 +771,7 @@ test("skips MCP resource requests when the capability is absent", async () => {
     Effect.scoped(
       Effect.gen(function* () {
         const server = yield* resourceServer({ resources: false })
-        const connection = yield* MCPClient.connect(
+        const connection = yield* connect(
           "resources",
           new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }),
           import.meta.dir,