refactor(core): spawn MCP servers through environment

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,6 +174,7 @@ 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
@@ -520,6 +522,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 +832,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,154 @@
+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, Deferred, Duration, Effect, Exit, 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)
+
+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 context = yield* Effect.context<Scope.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.unbounded<Uint8Array, Cause.Done>()
+  const encoder = new TextEncoder()
+  const buffer = new ReadBuffer()
+  const closed = Deferred.makeUnsafe<void>()
+  const state: { started: boolean; handle?: ChildProcessHandle } = { started: false }
+
+  const transport: Transport = {
+    start: () => {
+      if (state.started) return Promise.reject(new Error("Stdio transport already started"))
+      state.started = true
+      return Effect.runPromiseWith(context)(
+        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.fromQueue(outgoing), endOnDone: true },
+              stdout: "pipe",
+              stderr: "pipe",
+              forceKillAfter: FORCE_KILL_AFTER,
+            }),
+          )
+          state.handle = handle
+          yield* startOutput(handle)
+        }),
+      )
+    },
+    send: (message: JSONRPCMessage) =>
+      Queue.offerUnsafe(outgoing, encoder.encode(serializeMessage(message)))
+        ? Promise.resolve()
+        : Promise.reject(new Error("Not connected")),
+    close: () =>
+      Effect.runPromise(
+        Effect.gen(function* () {
+          Queue.endUnsafe(outgoing)
+          const handle = state.handle
+          if (!handle) return
+          // Give the server the same chance to exit on its own that the SDK transport gives it; the
+          // handle then signals the process through whichever execution driver spawned it.
+          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.ensuring(Effect.sync(() => buffer.clear())), Effect.ignore),
+      ),
+  }
+
+  const deliver = (chunk: Uint8Array) =>
+    Effect.gen(function* () {
+      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.sync(() => {
+              if (Deferred.doneUnsafe(closed, Exit.void)) transport.onclose?.()
+            }),
+          ),
+        ),
+      )
+
+      // StdioClientTransport pipes stderr into a stream nobody reads, which both hides server
+      // diagnostics and lets a chatty server stall on backpressure. Draining it into the debug log
+      // keeps the pipe moving and makes the output reachable.
+      yield* Effect.forkScoped(
+        handle.stderr.pipe(
+          Stream.decodeText(),
+          Stream.splitLines,
+          Stream.runForEach((line) =>
+            line.trim() === "" ? Effect.void : Effect.logDebug("mcp server stderr", { server: options.server, line }),
+          ),
+          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,6 +22,7 @@ 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"
@@ -31,10 +32,12 @@ 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 { ChildProcess } from "effect/unstable/process"
 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 +170,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 +233,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 +415,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 +448,41 @@ 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]
+  expect(command?._tag).toBe("StandardCommand")
+  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" })
+  expect(command.options.env).not.toHaveProperty("HOME")
+})
+
 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 +504,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 +525,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 +546,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 +565,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 +600,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 +671,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,