fix(core): tighten MCP transport shutdown

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/src/session/runner/llm.ts

@@ -113,8 +113,8 @@ const layer = Layer.effect(
     const compaction = yield* SessionCompaction.Service
     const title = yield* SessionTitle.Service
     const toolOutput = yield* ToolOutput.Service
-    // Title generation starts once input is visible and must not delay model execution.
-    // The in-flight set coalesces overlapping prompts while title presence records success durably.
+    // Title generation is a side effect of a successful step; it must not delay continuation.
+    // The in-flight set coalesces overlapping steps while title presence records success durably.
     const titlesRunning = new Set<SessionSchema.ID>()
     const forkTitle = yield* FiberSet.makeRuntime<never, void, never>()
     /**
@@ -144,6 +144,7 @@ const layer = Layer.effect(
       let step = 1
       while (true) {
         const result = yield* runStep(sessionID, promotable, step)
+        if (step === 1) yield* startTitle(sessionID)
         yield* runPendingCompaction(sessionID)
         if (!result.needsContinuation && !(yield* SessionPending.has(db, sessionID, "steer"))) return
         promotable = "steer"
@@ -235,7 +236,6 @@ const layer = Layer.effect(
       // a blocked first step leaves pending inputs untouched.
       yield* InstructionState.prepare(db, bus, selected.instructions, selected.session.id)
       const promoted = promotable ? yield* SessionPending.promote(db, bus, selected.session.id, promotable) : 0
-      if (promoted > 0) yield* startTitle(sessionID)
       // Promoted input opens a fresh step allowance.
       const currentStep = promoted > 0 ? 1 : step
       const loaded = yield* context.load(selected)

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,

packages/core/test/session-runner.test.ts

@@ -816,7 +816,7 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) =>
   })
 
 describe("SessionRunnerLLM", () => {
-  it.effect("generates the title while the first model step is still running", () =>
+  it.effect("retries title generation from the first prompt after execution and title failures", () =>
     Effect.gen(function* () {
       const session = yield* setup
       const agents = yield* Agent.Service
@@ -831,48 +831,16 @@ describe("SessionRunnerLLM", () => {
       )
 
       yield* admit(session, "First prompt")
-      yield* TestLLM.push(TestLLM.text("Generated title", "text-title"), Stream.never)
-      const bus = yield* Bus.Service
-      const renamed = yield* bus.subscribe(SessionEvent.Renamed).pipe(
-        Stream.filter((event) => event.data.sessionID === sessionID),
-        Stream.take(1),
-        Stream.runDrain,
-        Effect.forkScoped({ startImmediately: true }),
-      )
-      const runner = yield* SessionRunner.Service
-      const fiber = yield* runner.drain({ sessionID, force: true }).pipe(Effect.forkChild)
-      yield* Fiber.join(renamed)
-
-      expect((yield* session.get(sessionID)).title).toBe("Generated title")
-      yield* Fiber.interrupt(fiber)
-    }),
-  )
-
-  it.effect("retries title generation from the first prompt after title and execution failures", () =>
-    Effect.gen(function* () {
-      const session = yield* setup
-      const agents = yield* Agent.Service
-      const { db } = yield* Database.Service
-      yield* db.update(SessionTable).set({ title: null }).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie)
-      yield* agents.transform((draft) =>
-        draft.update(Agent.ID.make("title"), (agent) => {
-          agent.mode = "primary"
-          agent.hidden = true
-          agent.system = "Generate a title."
-        }),
-      )
-
-      yield* admit(session, "First prompt")
-      yield* TestLLM.push(Stream.fail(invalidRequest()), Stream.fail(invalidRequest()))
+      yield* TestLLM.push(Stream.fail(invalidRequest()))
       expect((yield* session.resume(sessionID).pipe(Effect.exit))._tag).toBe("Failure")
 
       yield* admit(session, "Second prompt")
       const titleFailed = yield* Deferred.make<void>()
       yield* TestLLM.push(
+        TestLLM.text("Recovered", "text-recovered"),
         Stream.make(LLMEvent.providerError({ message: "Title provider unavailable" })).pipe(
           Stream.ensuring(Deferred.succeed(titleFailed, undefined)),
         ),
-        TestLLM.text("Recovered", "text-recovered"),
       )
       yield* session.resume(sessionID)
       yield* Deferred.await(titleFailed)
@@ -888,13 +856,13 @@ describe("SessionRunnerLLM", () => {
       )
       yield* admit(session, "Third prompt")
       yield* TestLLM.push(
-        TestLLM.text("Generated title", "text-title"),
         TestLLM.text("Recovered again", "text-recovered-again"),
+        TestLLM.text("Generated title", "text-title"),
       )
       yield* session.resume(sessionID)
       yield* Fiber.join(renamed)
 
-      expect(requests).toHaveLength(6)
+      expect(requests).toHaveLength(5)
       expect(requests[2]?.messages).toContainEqual(Message.user("First prompt"))
       expect(requests[4]?.messages).toContainEqual(Message.user("First prompt"))
       expect((yield* session.get(sessionID)).title).toBe("Generated title")

packages/core/test/snapshot.test.ts

@@ -143,47 +143,44 @@ describe("Snapshot", () => {
     ),
   )
 
-  testEffect(Layer.empty).live(
-    "isolates snapshot indexes by canonical Git worktree",
-    () =>
-      Effect.acquireUseRelease(
-        Effect.promise(() => tmpdir()),
-        (tmp) =>
-          Effect.gen(function* () {
-            const project = path.join(tmp.path, "project")
-            const linked = path.join(tmp.path, "linked")
-            yield* Effect.promise(async () => {
-              await fs.mkdir(project)
-              await fs.writeFile(path.join(project, "tracked.txt"), "main\n")
-              await initGit(project, true)
-              await $`git -c core.fsmonitor=false worktree add --detach ${linked} HEAD`.cwd(project).quiet()
-            })
+  testEffect(Layer.empty).live("isolates snapshot indexes by canonical Git worktree", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => tmpdir()),
+      (tmp) =>
+        Effect.gen(function* () {
+          const project = path.join(tmp.path, "project")
+          const linked = path.join(tmp.path, "linked")
+          yield* Effect.promise(async () => {
+            await fs.mkdir(project)
+            await fs.writeFile(path.join(project, "tracked.txt"), "main\n")
+            await initGit(project, true)
+            await $`git -c core.fsmonitor=false worktree add --detach ${linked} HEAD`.cwd(project).quiet()
+          })
 
-            const capture = (directory: string) =>
-              Effect.gen(function* () {
-                const snapshot = yield* Snapshot.Service
-                return yield* snapshot.capture()
-              }).pipe(Effect.provide(snapshotLayer(tmp.path, directory)))
-            expect(yield* capture(project)).toBeDefined()
-            expect(yield* capture(linked)).toBeDefined()
+          const capture = (directory: string) =>
+            Effect.gen(function* () {
+              const snapshot = yield* Snapshot.Service
+              return yield* snapshot.capture()
+            }).pipe(Effect.provide(snapshotLayer(tmp.path, directory)))
+          expect(yield* capture(project)).toBeDefined()
+          expect(yield* capture(linked)).toBeDefined()
 
-            const projectID = yield* Effect.gen(function* () {
-              return (yield* Location.Service).project.id
-            }).pipe(
-              Effect.provide(
-                AppNodeBuilder.build(Location.boundNode(Location.Ref.make({ directory: AbsolutePath.make(project) }))),
-              ),
-            )
-            expect(
-              yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(project)))),
-            ).toBeDefined()
-            expect(
-              yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(linked)))),
-            ).toBeDefined()
-          }),
-        (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
-      ),
-    { timeout: 15_000 },
+          const projectID = yield* Effect.gen(function* () {
+            return (yield* Location.Service).project.id
+          }).pipe(
+            Effect.provide(
+              AppNodeBuilder.build(Location.boundNode(Location.Ref.make({ directory: AbsolutePath.make(project) }))),
+            ),
+          )
+          expect(
+            yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(project)))),
+          ).toBeDefined()
+          expect(
+            yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(linked)))),
+          ).toBeDefined()
+        }),
+      (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+    ),
   )
 })
 

packages/core/test/tool-shell.test.ts

@@ -387,63 +387,57 @@ describe("ShellTool", () => {
     ),
   )
 
-  it.live(
-    "approves an explicit external workdir before shell execution",
-    () =>
-      Effect.acquireUseRelease(
-        Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
-        ([active, outside]) => {
-          reset()
-          return withSession(active.path, (registry) =>
-            executeTool(registry, call({ command: cwdCommand, workdir: outside.path })),
-          ).pipe(
-            Effect.andThen(
-              Effect.sync(() => {
-                expect(assertions.map((item) => item.action)).toEqual(["external_directory", "shell"])
-                expect(assertions[0]).toMatchObject({
-                  resources: [path.join(realpathSync(outside.path), "*").replaceAll("\\", "/")],
-                })
-              }),
-            ),
-          )
-        },
-        ([active, outside]) =>
-          Effect.promise(() =>
-            Promise.all([active[Symbol.asyncDispose](), outside[Symbol.asyncDispose]()]).then(() => undefined),
+  it.live("approves an explicit external workdir before shell execution", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
+      ([active, outside]) => {
+        reset()
+        return withSession(active.path, (registry) =>
+          executeTool(registry, call({ command: cwdCommand, workdir: outside.path })),
+        ).pipe(
+          Effect.andThen(
+            Effect.sync(() => {
+              expect(assertions.map((item) => item.action)).toEqual(["external_directory", "shell"])
+              expect(assertions[0]).toMatchObject({
+                resources: [path.join(realpathSync(outside.path), "*").replaceAll("\\", "/")],
+              })
+            }),
           ),
-      ),
-    { timeout: 15_000 },
+        )
+      },
+      ([active, outside]) =>
+        Effect.promise(() =>
+          Promise.all([active[Symbol.asyncDispose](), outside[Symbol.asyncDispose]()]).then(() => undefined),
+        ),
+    ),
   )
 
-  it.live(
-    "approves an external directory used by a directory-change command",
-    () =>
-      Effect.acquireUseRelease(
-        Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
-        ([active, outside]) => {
-          reset()
-          const command = isWindows
-            ? `Set-Location -LiteralPath '${outside.path}'; (Get-Location).Path`
-            : `cd '${outside.path}' && pwd`
-          return withSession(active.path, (registry) =>
-            executeTool(registry, call({ command }, "call-external-cd")),
-          ).pipe(
-            Effect.andThen(
-              Effect.sync(() => {
-                expect(assertions.map((item) => item.action)).toEqual(["external_directory", "shell"])
-                expect(assertions[0]).toMatchObject({
-                  resources: [path.join(realpathSync(outside.path), "*").replaceAll("\\", "/")],
-                })
-              }),
-            ),
-          )
-        },
-        ([active, outside]) =>
-          Effect.promise(() =>
-            Promise.all([active[Symbol.asyncDispose](), outside[Symbol.asyncDispose]()]).then(() => undefined),
+  it.live("approves an external directory used by a directory-change command", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
+      ([active, outside]) => {
+        reset()
+        const command = isWindows
+          ? `Set-Location -LiteralPath '${outside.path}'; (Get-Location).Path`
+          : `cd '${outside.path}' && pwd`
+        return withSession(active.path, (registry) =>
+          executeTool(registry, call({ command }, "call-external-cd")),
+        ).pipe(
+          Effect.andThen(
+            Effect.sync(() => {
+              expect(assertions.map((item) => item.action)).toEqual(["external_directory", "shell"])
+              expect(assertions[0]).toMatchObject({
+                resources: [path.join(realpathSync(outside.path), "*").replaceAll("\\", "/")],
+              })
+            }),
           ),
-      ),
-    { timeout: 15_000 },
+        )
+      },
+      ([active, outside]) =>
+        Effect.promise(() =>
+          Promise.all([active[Symbol.asyncDispose](), outside[Symbol.asyncDispose]()]).then(() => undefined),
+        ),
+    ),
   )
 
   it.live("approves an expanded external home directory", () =>
@@ -465,31 +459,28 @@ describe("ShellTool", () => {
     ),
   )
 
-  it.live(
-    "does not execute after external-directory or shell denial",
-    () =>
-      Effect.acquireUseRelease(
-        Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
-        ([active, outside]) =>
-          Effect.gen(function* () {
-            reset()
-            denyAction = "external_directory"
-            yield* withSession(active.path, (registry) =>
-              executeTool(registry, call({ command: cwdCommand, workdir: outside.path })),
-            )
-            expect(assertions.map((item) => item.action)).toEqual(["external_directory"])
+  it.live("does not execute after external-directory or shell denial", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
+      ([active, outside]) =>
+        Effect.gen(function* () {
+          reset()
+          denyAction = "external_directory"
+          yield* withSession(active.path, (registry) =>
+            executeTool(registry, call({ command: cwdCommand, workdir: outside.path })),
+          )
+          expect(assertions.map((item) => item.action)).toEqual(["external_directory"])
 
-            reset()
-            denyAction = "shell"
-            yield* withSession(active.path, (registry) => executeTool(registry, call({ command: cwdCommand })))
-            expect(assertions.map((item) => item.action)).toEqual(["shell"])
-          }),
-        ([active, outside]) =>
-          Effect.promise(() =>
-            Promise.all([active[Symbol.asyncDispose](), outside[Symbol.asyncDispose]()]).then(() => undefined),
-          ),
-      ),
-    { timeout: 15_000 },
+          reset()
+          denyAction = "shell"
+          yield* withSession(active.path, (registry) => executeTool(registry, call({ command: cwdCommand })))
+          expect(assertions.map((item) => item.action)).toEqual(["shell"])
+        }),
+      ([active, outside]) =>
+        Effect.promise(() =>
+          Promise.all([active[Symbol.asyncDispose](), outside[Symbol.asyncDispose]()]).then(() => undefined),
+        ),
+    ),
   )
 
   it.live("keeps non-zero exits useful", () =>
@@ -628,7 +619,7 @@ describe("ShellTool", () => {
         },
         (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]().then(() => undefined)),
       ),
-    { timeout: 15_000 },
+    { timeout: 10_000 },
   )
 
   it.live(
@@ -639,7 +630,7 @@ describe("ShellTool", () => {
         (tmp) => {
           reset()
           return withSession(tmp.path, (registry) =>
-            executeTool(registry, call({ command: timeoutOutputCommand, timeout: isWindows ? 3_000 : 500 })),
+            executeTool(registry, call({ command: timeoutOutputCommand, timeout: isWindows ? 3_000 : 50 })),
           ).pipe(
             Effect.andThen((settled) =>
               Effect.sync(() => {

packages/tui/test/component/tab-pulse.test.tsx

@@ -1,6 +1,10 @@
+/** @jsxImportSource @opentui/solid */
 import { expect, test } from "bun:test"
 import { RGBA } from "@opentui/core"
+import { testRender } from "@opentui/solid"
+import { createSignal } from "solid-js"
 import {
+  TabPulse,
   blendTabPulseColor,
   completionPulseOpacity,
   glowIgnitionLevel,
@@ -8,6 +12,50 @@ import {
 } from "../../src/component/tab-pulse"
 import { tint } from "../../src/theme/color"
 
+test("a prompt pulse restarts the neutral edge flash while the tab remains busy", async () => {
+  const background = RGBA.fromHex("#101010")
+  const flash = RGBA.fromHex("#f0f0f0")
+  const [promptPulse, setPromptPulse] = createSignal(0)
+  const app = await testRender(
+    () => (
+      <box width={8} height={1} backgroundColor={background}>
+        <TabPulse
+          active={true}
+          promptPulse={promptPulse()}
+          color={background}
+          flashColor={flash}
+          backgroundColor={background}
+        />
+      </box>
+    ),
+    { width: 8, height: 1 },
+  )
+
+  const firstBackground = () => app.captureSpans().lines[0]?.spans[0]?.bg
+
+  try {
+    await app.renderOnce()
+    expect(firstBackground()?.equals(background)).toBeTrue()
+
+    setPromptPulse(1)
+    await Bun.sleep(80)
+    await app.renderOnce()
+    expect(firstBackground()?.equals(background)).toBeFalse()
+    expect(firstBackground()?.r ?? 0).toBeGreaterThan(0.17)
+
+    await Bun.sleep(800)
+    await app.renderOnce()
+    expect(firstBackground()?.equals(background)).toBeTrue()
+
+    setPromptPulse(2)
+    await Bun.sleep(80)
+    await app.renderOnce()
+    expect(firstBackground()?.equals(background)).toBeFalse()
+  } finally {
+    app.renderer.destroy()
+  }
+})
+
 test("completion pulse rises quickly and fades over the remaining duration", () => {
   expect(completionPulseOpacity(0)).toBe(0)
   expect(completionPulseOpacity(0.06)).toBeCloseTo(0.5)