fix(core): settle owned process output

packages/core/src/cross-spawn-spawner.ts

@@ -24,6 +24,7 @@ import {
 import * as NodeChildProcess from "node:child_process"
 import { PassThrough } from "node:stream"
 import launch from "cross-spawn"
+import { ProcessOutput } from "./process-output"
 
 const toError = (err: unknown): Error => (err instanceof globalThis.Error ? err : new globalThis.Error(String(err)))
 
@@ -266,18 +267,11 @@ export const make = Effect.gen(function* () {
     Effect.callback<readonly [NodeChildProcess.ChildProcess, ExitSignal], PlatformError.PlatformError>((resume) => {
       const signal = Deferred.makeUnsafe<readonly [code: number | null, signal: NodeJS.Signals | null]>()
       const proc = launch(command.command, command.args, opts)
-      let end = false
-      let exit: readonly [code: number | null, signal: NodeJS.Signals | null] | undefined
       proc.on("error", (err) => {
         resume(Effect.fail(toPlatformError("spawn", err, command)))
       })
       proc.on("exit", (...args) => {
-        exit = args
-      })
-      proc.on("close", (...args) => {
-        if (end) return
-        end = true
-        Deferred.doneUnsafe(signal, Exit.succeed(exit ?? args))
+        Deferred.doneUnsafe(signal, Exit.succeed(args))
       })
       proc.on("spawn", () => {
         resume(Effect.succeed([proc, signal]))
@@ -340,6 +334,16 @@ export const make = Effect.gen(function* () {
       })
     }
 
+  const groupAlive = (proc: NodeChildProcess.ChildProcess) => {
+    if (process.platform === "win32") return false
+    try {
+      process.kill(-proc.pid!, 0)
+      return true
+    } catch {
+      return false
+    }
+  }
+
   const source = (handle: ChildProcessHandle, from: ChildProcess.PipeFromOption | undefined) => {
     const opt = from ?? "stdout"
     switch (opt) {
@@ -381,9 +385,11 @@ export const make = Effect.gen(function* () {
               const done = yield* Deferred.isDone(signal)
               const kill = timeout(proc, command, command.options)
               if (done) {
-                const [code] = yield* Deferred.await(signal)
                 if (process.platform === "win32") return yield* Effect.void
-                if (code !== 0 && Predicate.isNotNull(code)) return yield* Effect.ignore(kill(killGroup))
+                if (!groupAlive(proc)) return yield* Effect.void
+                yield* Effect.ignore(killGroup(command, proc, command.options.killSignal ?? "SIGTERM"))
+                yield* Effect.sleep("100 millis")
+                if (groupAlive(proc)) yield* Effect.ignore(killGroup(command, proc, "SIGKILL"))
                 return yield* Effect.void
               }
               const send = (s: NodeJS.Signals) =>
@@ -403,7 +409,7 @@ export const make = Effect.gen(function* () {
           const fd = yield* setupFds(command, proc, extra)
           const out = setupOutput(command, proc, sout, serr)
           let ref = true
-          return makeHandle({
+          const handle = makeHandle({
             pid: ProcessId(proc.pid!),
             stdin: yield* setupStdin(command, proc, sin),
             stdout: out.stdout,
@@ -446,6 +452,8 @@ export const make = Effect.gen(function* () {
               })
             }),
           })
+          ProcessOutput.register(handle, proc)
+          return handle
         }
         case "PipedCommand": {
           const flat = flatten(command)

packages/core/src/process-output.ts

@@ -0,0 +1,68 @@
+import { Cause, Deferred, Duration, Effect, Exit, Fiber } from "effect"
+import type { ChildProcess } from "node:child_process"
+import type { ChildProcessHandle } from "effect/unstable/process/ChildProcessSpawner"
+
+const processes = new WeakMap<ChildProcessHandle, ChildProcess>()
+
+export const register = (handle: ChildProcessHandle, process: ChildProcess) => {
+  processes.set(handle, process)
+}
+
+export const drain = <E, R>(
+  handle: ChildProcessHandle,
+  drains: ReadonlyArray<Effect.Effect<unknown, E, R>>,
+  options?: { readonly grace?: Duration.Input; readonly onClose?: () => void },
+) =>
+  Effect.uninterruptibleMask((restore) =>
+    Effect.gen(function* () {
+      const fibers = yield* Effect.forEach(drains, (drain) => Effect.forkDetach(drain))
+      let closed = false
+      const close = Effect.sync(() => {
+        if (closed) return
+        closed = true
+        options?.onClose?.()
+        const process = processes.get(handle)
+        process?.stdout?.destroy()
+        process?.stderr?.destroy()
+        for (const stream of process?.stdio.slice(3) ?? []) {
+          if (stream && "readable" in stream && stream.readable) stream.destroy()
+        }
+        for (const fiber of fibers) fiber.interruptUnsafe()
+      })
+      const failed = yield* Deferred.make<Cause.Cause<E>>()
+      const observers = fibers.map((fiber) =>
+        fiber.addObserver((exit) => {
+          if (Exit.isFailure(exit)) Deferred.doneUnsafe(failed, Effect.succeed(exit.cause))
+        }),
+      )
+      const run = Effect.gen(function* () {
+        const outcome = yield* Effect.raceAllFirst([
+          handle.exitCode.pipe(
+            Effect.exit,
+            Effect.map((exit) => ({ type: "exit" as const, exit })),
+          ),
+          Deferred.await(failed).pipe(Effect.map((cause) => ({ type: "failure" as const, cause }))),
+        ])
+        for (const remove of observers) remove()
+        if (outcome.type === "failure") {
+          yield* close
+          yield* handle.kill({ forceKillAfter: "1 second" }).pipe(Effect.ignore)
+          return yield* Effect.failCause(outcome.cause)
+        }
+
+        const exits = yield* Effect.forEach(fibers, (fiber) => Fiber.await(fiber), { concurrency: "unbounded" }).pipe(
+          Effect.timeoutOrElse({ duration: options?.grace ?? "1 second", orElse: () => Effect.succeed(undefined) }),
+        )
+        if (exits) {
+          const failure = exits.find((exit) => Exit.isFailure(exit))
+          if (failure) return yield* Effect.failCause(failure.cause)
+        } else {
+          yield* close
+        }
+        return Exit.isFailure(outcome.exit) ? yield* Effect.failCause(outcome.exit.cause) : outcome.exit.value
+      })
+      return yield* restore(run).pipe(Effect.onInterrupt(() => close))
+    }),
+  )
+
+export * as ProcessOutput from "./process-output"

packages/core/src/process.ts

@@ -3,6 +3,7 @@ import type { PlatformError } from "effect/PlatformError"
 import { ChildProcess } from "effect/unstable/process"
 import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
 import { CrossSpawnSpawner } from "./cross-spawn-spawner"
+import { ProcessOutput } from "./process-output"
 
 export class AppProcessError extends Schema.TaggedErrorClass<AppProcessError>()("AppProcessError", {
   command: Schema.String,
@@ -125,6 +126,19 @@ export const collectStream = (stream: Stream.Stream<Uint8Array, PlatformError>,
     },
   ).pipe(Effect.map((x) => ({ buffer: Buffer.concat(x.chunks), truncated: x.truncated })))
 
+const collector = (stream: Stream.Stream<Uint8Array, PlatformError>, maxBytes: number | undefined) => {
+  const state = { chunks: [] as Uint8Array[], bytes: 0, truncated: false }
+  const drain = Stream.runForEach(stream, (chunk) =>
+    Effect.sync(() => {
+      const remaining = maxBytes === undefined ? chunk.length : maxBytes - state.bytes
+      if (remaining > 0) state.chunks.push(remaining >= chunk.length ? chunk : chunk.slice(0, remaining))
+      state.bytes += chunk.length
+      state.truncated = maxBytes !== undefined && state.bytes > maxBytes
+    }),
+  )
+  return { drain, result: () => ({ buffer: Buffer.concat(state.chunks), truncated: state.truncated }) }
+}
+
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
@@ -135,21 +149,18 @@ export const layer = Layer.effect(
       const collect = Effect.scoped(
         Effect.gen(function* () {
           const handle = yield* spawner.spawn(command)
-          const [stdout, stderr, exitCode] = yield* Effect.all(
-            [
-              collectStream(handle.stdout, options?.maxOutputBytes),
-              collectStream(handle.stderr, options?.maxErrorBytes),
-              handle.exitCode,
-            ],
-            { concurrency: "unbounded" },
-          )
+          const stdout = collector(handle.stdout, options?.maxOutputBytes)
+          const stderr = collector(handle.stderr, options?.maxErrorBytes)
+          const exitCode = yield* ProcessOutput.drain(handle, [stdout.drain, stderr.drain])
+          const out = stdout.result()
+          const err = stderr.result()
           return {
             command: description,
             exitCode,
-            stdout: stdout.buffer,
-            stderr: stderr.buffer,
-            stdoutTruncated: stdout.truncated,
-            stderrTruncated: stderr.truncated,
+            stdout: out.buffer,
+            stderr: err.buffer,
+            stdoutTruncated: out.truncated,
+            stderrTruncated: err.truncated,
           } satisfies RunResult
         }),
       )

packages/core/test/effect/cross-spawn-spawner.test.ts

@@ -6,6 +6,7 @@ import { Effect, Exit, Stream } from "effect"
 import type * as PlatformError from "effect/PlatformError"
 import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
 import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
+import { ProcessOutput } from "@opencode-ai/core/process-output"
 import { testEffect } from "../lib/effect"
 
 const live = CrossSpawnSpawner.defaultLayer
@@ -96,6 +97,28 @@ describe("cross-spawn spawner", () => {
         expect(code).toBe(ChildProcessSpawner.ExitCode(42))
       }),
     )
+
+    fx.live(
+      "reports direct exit while a descendant holds stdout open",
+      Effect.gen(function* () {
+        if (process.platform === "win32") return
+        const handle = yield* ChildProcess.make("sh", ["-c", "sleep 30 &"])
+        expect(yield* handle.exitCode.pipe(Effect.timeout("1 second"))).toBe(ChildProcessSpawner.ExitCode(0))
+      }),
+    )
+
+    fx.live(
+      "terminates the process when an owned drain fails before exit",
+      Effect.gen(function* () {
+        const handle = yield* js("setInterval(() => {}, 10_000)")
+        const pid = Number(handle.pid)
+        const exit = yield* ProcessOutput.drain(handle, [Effect.fail("drain failed")]).pipe(Effect.exit)
+
+        expect(Exit.isFailure(exit)).toBe(true)
+        expect(yield* Effect.promise(() => gone(pid))).toBe(true)
+      }),
+      5_000,
+    )
   })
 
   describe("cwd option", () => {

packages/core/test/process/process.test.ts

@@ -25,6 +25,19 @@ const waitForFile = (file: string) =>
     }
   })
 
+const gone = (pid: number) =>
+  Effect.promise(async () => {
+    for (let i = 0; i < 200; i++) {
+      try {
+        process.kill(pid, 0)
+      } catch {
+        return true
+      }
+      await new Promise<void>((resolve) => setTimeout(resolve, 25))
+    }
+    return false
+  })
+
 describe("AppProcess", () => {
   describe("run", () => {
     it.effect(
@@ -134,6 +147,22 @@ describe("AppProcess", () => {
     )
 
     if (process.platform !== "win32") {
+      it.live(
+        "captures large output and cleans a descendant holding stdout open",
+        Effect.gen(function* () {
+          const svc = yield* AppProcess.Service
+          const size = 8 * 1024 * 1024
+          const command = `sleep 30 & child=$!; printf "%s\\n" "$child"; dd if=/dev/zero bs=${size} count=1 2>/dev/null | tr '\\0' x`
+          const result = yield* svc.run(ChildProcess.make("sh", ["-c", command]))
+          const newline = result.stdout.indexOf(10)
+          const pid = Number(result.stdout.subarray(0, newline).toString("utf8"))
+
+          expect(result.stdout.length).toBe(newline + 1 + size)
+          expect(yield* gone(pid)).toBe(true)
+        }),
+        8_000,
+      )
+
       it.live(
         "timeout cleans up the scoped child process",
         Effect.acquireUseRelease(

packages/opencode/src/session/prompt.ts

@@ -23,6 +23,7 @@ import { LSP } from "@/lsp/lsp"
 import { ulid } from "ulid"
 import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
 import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
+import { ProcessOutput } from "@opencode-ai/core/process-output"
 import * as Stream from "effect/Stream"
 import { Command } from "../command"
 import { pathToFileURL, fileURLToPath } from "url"
@@ -563,38 +564,48 @@ export const layer = Layer.effect(
           const args = Shell.args(sh, input.command, cwd)
           let output = ""
           let aborted = false
+          let acceptingOutput = true
 
-          const finish = Effect.uninterruptible(
-            Effect.gen(function* () {
-              if (aborted) {
-                output += "\n\n" + ["<metadata>", "User aborted the command", "</metadata>"].join("\n")
-              }
-              const completed = Date.now()
-              if (flags.experimentalEventSystem) {
-                yield* events.publish(SessionEvent.Shell.Ended, {
-                  sessionID: input.sessionID,
-                  timestamp: DateTime.makeUnsafe(completed),
-                  callID: part.callID,
-                  output,
-                })
-              }
-              if (!msg.time.completed) {
-                msg.time.completed = completed
-                yield* sessions.updateMessage(msg)
-              }
-              if (part.state.status === "running") {
-                part.state = {
-                  status: "completed",
-                  time: { ...part.state.time, end: completed },
-                  input: part.state.input,
-                  title: "",
-                  metadata: { output, description: "" },
-                  output,
+          const finish = (error?: string) =>
+            Effect.uninterruptible(
+              Effect.gen(function* () {
+                if (aborted) {
+                  output += "\n\n" + ["<metadata>", "User aborted the command", "</metadata>"].join("\n")
                 }
-                yield* sessions.updatePart(part)
-              }
-            }),
-          )
+                const completed = Date.now()
+                if (flags.experimentalEventSystem) {
+                  yield* events.publish(SessionEvent.Shell.Ended, {
+                    sessionID: input.sessionID,
+                    timestamp: DateTime.makeUnsafe(completed),
+                    callID: part.callID,
+                    output,
+                  })
+                }
+                if (!msg.time.completed) {
+                  msg.time.completed = completed
+                  yield* sessions.updateMessage(msg)
+                }
+                if (part.state.status === "running") {
+                  part.state = error
+                    ? {
+                        status: "error",
+                        error,
+                        time: { ...part.state.time, end: completed },
+                        input: part.state.input,
+                        metadata: { output, description: "" },
+                      }
+                    : {
+                        status: "completed",
+                        time: { ...part.state.time, end: completed },
+                        input: part.state.input,
+                        title: "",
+                        metadata: { output, description: "" },
+                        output,
+                      }
+                  yield* sessions.updatePart(part)
+                }
+              }),
+            )
 
           const exit = yield* restore(
             Effect.gen(function* () {
@@ -611,27 +622,39 @@ export const layer = Layer.effect(
                 forceKillAfter: "3 seconds",
               })
               const handle = yield* spawner.spawn(cmd)
-              yield* Stream.runForEach(Stream.decodeText(handle.all), (chunk) =>
-                Effect.gen(function* () {
-                  output += chunk
-                  if (part.state.status === "running") {
-                    part.state.metadata = { output, description: "" }
-                    yield* sessions.updatePart(part)
-                  }
-                }),
-              )
-              yield* handle.exitCode
+              yield* ProcessOutput.drain(
+                handle,
+                [
+                  Stream.runForEach(Stream.decodeText(handle.all), (chunk) =>
+                    Effect.gen(function* () {
+                      if (!acceptingOutput) return
+                      output += chunk
+                      if (part.state.status === "running") {
+                        part.state.metadata = { output, description: "" }
+                        yield* sessions.updatePart(part)
+                      }
+                    }),
+                  ),
+                ],
+                {
+                  grace: "250 millis",
+                  onClose: () => {
+                    acceptingOutput = false
+                  },
+                },
+              ).pipe(Effect.asVoid)
             }).pipe(Effect.scoped, Effect.orDie),
           ).pipe(Effect.exit)
 
           if (Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause) && !Cause.hasDies(exit.cause)) {
             aborted = true
           }
-          yield* finish
+          const failure =
+            Exit.isFailure(exit) && !aborted && !Cause.hasInterruptsOnly(exit.cause) ? exit.cause : undefined
+          const error = failure && Cause.squash(failure)
+          yield* finish(error instanceof Error ? error.message : error ? String(error) : undefined)
 
-          if (Exit.isFailure(exit) && !aborted && !Cause.hasInterruptsOnly(exit.cause)) {
-            return yield* Effect.failCause(exit.cause)
-          }
+          if (failure) return yield* Effect.failCause(failure)
 
           return { info: msg, parts: [part] }
         }),

packages/opencode/src/tool/shell.ts

@@ -1,4 +1,4 @@
-import { Effect, Stream } from "effect"
+import { Effect, Fiber, Stream } from "effect"
 import os from "os"
 import { createWriteStream } from "node:fs"
 import * as Tool from "./tool"
@@ -20,6 +20,7 @@ import * as Truncate from "./truncate"
 import { Plugin } from "@/plugin"
 import { ChildProcess } from "effect/unstable/process"
 import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
+import { ProcessOutput } from "@opencode-ai/core/process-output"
 import { ShellPrompt, type Parameters } from "./shell/prompt"
 import { BashArity } from "@/permission/arity"
 
@@ -456,6 +457,7 @@ export const ShellTool = Tool.define(
       let cut = false
       let expired = false
       let aborted = false
+      let acceptingOutput = true
 
       const closeSink = Effect.fnUntraced(function* () {
         const stream = sink
@@ -494,54 +496,59 @@ export const ShellTool = Tool.define(
           yield* Effect.addFinalizer(closeSink)
           const handle = yield* spawner.spawn(cmd(input.shell, input.command, input.cwd, input.env))
 
-          yield* Effect.forkScoped(
-            Stream.runForEach(Stream.decodeText(handle.all), (chunk) => {
-              const size = Buffer.byteLength(chunk, "utf-8")
-              list.push({ text: chunk, size })
-              used += size
-              while (used > keep && list.length > 1) {
-                const item = list.shift()
-                if (!item) break
-                used -= item.size
-                cut = true
+          const output = Stream.runForEach(Stream.decodeText(handle.all), (chunk) => {
+            if (!acceptingOutput) return Effect.void
+            const size = Buffer.byteLength(chunk, "utf-8")
+            list.push({ text: chunk, size })
+            used += size
+            while (used > keep && list.length > 1) {
+              const item = list.shift()
+              if (!item) break
+              used -= item.size
+              cut = true
+            }
+
+            last = preview(last + chunk)
+
+            if (file) {
+              sink?.write(chunk)
+            } else {
+              full += chunk
+              if (Buffer.byteLength(full, "utf-8") > limits.maxBytes) {
+                return trunc.write(full).pipe(
+                  Effect.andThen((next) =>
+                    Effect.sync(() => {
+                      file = next
+                      cut = true
+                      sink = createWriteStream(next, { flags: "a" })
+                      full = ""
+                    }),
+                  ),
+                  Effect.andThen(
+                    ctx.metadata({
+                      metadata: {
+                        output: last,
+                        description: input.description,
+                      },
+                    }),
+                  ),
+                )
               }
+            }
 
-              last = preview(last + chunk)
-
-              if (file) {
-                sink?.write(chunk)
-              } else {
-                full += chunk
-                if (Buffer.byteLength(full, "utf-8") > limits.maxBytes) {
-                  return trunc.write(full).pipe(
-                    Effect.andThen((next) =>
-                      Effect.sync(() => {
-                        file = next
-                        cut = true
-                        sink = createWriteStream(next, { flags: "a" })
-                        full = ""
-                      }),
-                    ),
-                    Effect.andThen(
-                      ctx.metadata({
-                        metadata: {
-                          output: last,
-                          description: input.description,
-                        },
-                      }),
-                    ),
-                  )
-                }
-              }
-
-              return ctx.metadata({
-                metadata: {
-                  output: last,
-                  description: input.description,
-                },
-              })
-            }),
-          )
+            return ctx.metadata({
+              metadata: {
+                output: last,
+                description: input.description,
+              },
+            })
+          })
+          const drained = yield* ProcessOutput.drain(handle, [output], {
+            grace: "1500 millis",
+            onClose: () => {
+              acceptingOutput = false
+            },
+          }).pipe(Effect.forkScoped)
 
           const abort = Effect.callback<void>((resume) => {
             if (ctx.abort.aborted) return resume(Effect.void)
@@ -552,8 +559,8 @@ export const ShellTool = Tool.define(
 
           const timeout = Effect.sleep(`${input.timeout + 100} millis`)
 
-          const exit = yield* Effect.raceAll([
-            handle.exitCode.pipe(Effect.map((code) => ({ kind: "exit" as const, code }))),
+          const exit = yield* Effect.raceAllFirst([
+            Fiber.join(drained).pipe(Effect.map((code) => ({ kind: "exit" as const, code }))),
             abort.pipe(Effect.map(() => ({ kind: "abort" as const, code: null }))),
             timeout.pipe(Effect.map(() => ({ kind: "timeout" as const, code: null }))),
           ])
@@ -566,6 +573,7 @@ export const ShellTool = Tool.define(
             expired = true
             yield* handle.kill({ forceKillAfter: "3 seconds" }).pipe(Effect.orDie)
           }
+          if (exit.kind !== "exit") yield* Fiber.await(drained)
 
           return exit.kind === "exit" ? exit.code : null
         }),

packages/opencode/test/session/prompt.test.ts

@@ -6,7 +6,15 @@ import { eq } from "drizzle-orm"
 import { EventV2Bridge } from "@/event-v2-bridge"
 import { FetchHttpClient } from "effect/unstable/http"
 import { expect } from "bun:test"
-import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect"
+import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Sink, Stream } from "effect"
+import * as PlatformError from "effect/PlatformError"
+import {
+  ChildProcessSpawner,
+  ExitCode,
+  make as makeSpawner,
+  makeHandle,
+  ProcessId,
+} from "effect/unstable/process/ChildProcessSpawner"
 import path from "path"
 import { fileURLToPath, pathToFileURL } from "url"
 import { NamedError } from "@opencode-ai/core/util/error"
@@ -157,8 +165,6 @@ const lsp = Layer.succeed(
 
 const status = SessionStatus.layer.pipe(Layer.provideMerge(EventV2Bridge.defaultLayer))
 const run = SessionRunState.layer.pipe(Layer.provide(status))
-const infra = Layer.mergeAll(NodeFileSystem.layer, CrossSpawnSpawner.defaultLayer)
-
 const processorCreateStarted: Array<() => void> = []
 const blockingProcessor = Layer.succeed(
   SessionProcessor.Service,
@@ -167,7 +173,8 @@ const blockingProcessor = Layer.succeed(
   }),
 )
 
-function makePrompt(input?: { processor?: "blocking" }) {
+function makePrompt(input?: { processor?: "blocking"; spawner?: Layer.Layer<ChildProcessSpawner> }) {
+  const infra = Layer.mergeAll(NodeFileSystem.layer, input?.spawner ?? CrossSpawnSpawner.defaultLayer)
   const deps = Layer.mergeAll(
     Session.defaultLayer,
     Snapshot.defaultLayer,
@@ -236,17 +243,39 @@ function makePrompt(input?: { processor?: "blocking" }) {
   )
 }
 
-function makeHttp(input?: { processor?: "blocking" }) {
+function makeHttp(input?: { processor?: "blocking"; spawner?: Layer.Layer<ChildProcessSpawner> }) {
   return Layer.mergeAll(TestLLMServer.layer, makePrompt(input))
 }
 
-function makeHttpNoLLMServer(input?: { processor?: "blocking" }) {
+function makeHttpNoLLMServer(input?: { processor?: "blocking"; spawner?: Layer.Layer<ChildProcessSpawner> }) {
   return makePrompt(input)
 }
 
 const it = testEffect(makeHttp())
 const noLLMServer = testEffect(makeHttpNoLLMServer())
 const raceNoLLMServer = testEffect(makeHttpNoLLMServer({ processor: "blocking" }))
+const streamError = PlatformError.systemError({ _tag: "Unknown", module: "ChildProcess", method: "stdout" })
+const failingHandle = makeHandle({
+  pid: ProcessId(1),
+  stdin: Sink.drain,
+  stdout: Stream.fail(streamError),
+  stderr: Stream.empty,
+  all: Stream.fail(streamError),
+  getInputFd: () => Sink.drain,
+  getOutputFd: () => Stream.empty,
+  isRunning: Effect.succeed(false),
+  exitCode: Effect.succeed(ExitCode(0)),
+  kill: () => Effect.void,
+  unref: Effect.succeed(Effect.void),
+})
+const failingShell = testEffect(
+  makeHttpNoLLMServer({
+    spawner: Layer.succeed(
+      ChildProcessSpawner,
+      makeSpawner(() => Effect.succeed(failingHandle)),
+    ),
+  }),
+)
 const unix = process.platform !== "win32" ? it.instance : it.instance.skip
 const unixNoLLMServer = process.platform !== "win32" ? noLLMServer.instance : noLLMServer.instance.skip
 
@@ -1643,6 +1672,21 @@ unixNoLLMServer(
   30_000,
 )
 
+failingShell.instance(
+  "legacy shell persists error when output draining fails",
+  () =>
+    Effect.gen(function* () {
+      const { prompt, sessions, chat } = yield* boot()
+      const exit = yield* prompt.shell({ sessionID: chat.id, agent: "build", command: "ignored" }).pipe(Effect.exit)
+      const messages = yield* sessions.messages({ sessionID: chat.id })
+      const tool = messages.flatMap((message) => message.parts).find((part) => part.type === "tool")
+
+      expect(Exit.isFailure(exit)).toBe(true)
+      expect(tool?.state.status).toBe("error")
+    }),
+  { config: cfg },
+)
+
 unixNoLLMServer(
   "cancel persists aborted shell result when shell ignores TERM",
   () =>
@@ -1744,6 +1788,35 @@ unix(
   30_000,
 )
 
+unix(
+  "bash tool settles when a descendant inherits stdout",
+  () =>
+    Effect.gen(function* () {
+      const { dir, llm } = yield* useServerConfig(providerCfg)
+      const prompt = yield* SessionPrompt.Service
+      const sessions = yield* Session.Service
+      const chat = yield* sessions.create({ permission: [{ permission: "*", pattern: "*", action: "allow" }] })
+      yield* prompt.prompt({
+        sessionID: chat.id,
+        agent: "build",
+        noReply: true,
+        parts: [{ type: "text", text: "run" }],
+      })
+      yield* llm.tool("bash", {
+        command: "sleep 30 & printf done",
+        description: "background",
+        timeout: 30_000,
+        workdir: dir,
+      })
+      yield* llm.text("continued")
+
+      const result = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.timeout("3 seconds"))
+      expect(result.parts.some((part) => part.type === "text" && part.text === "continued")).toBe(true)
+    }),
+  { git: true },
+  8_000,
+)
+
 unixNoLLMServer(
   "cancel interrupts loop queued behind shell",
   () =>