refactor(test/cli): migrate serve/acp builders to AppProcess.spawn

Slice 2 of the CLI harness Effect migration. Drops the last raw
Bun.spawn call sites in withCliFixture.

- `serve` and `acp` both move from `Effect.acquireRelease(Bun.spawn(...))`
  to `appProc.spawn(ChildProcess.make(...))`. The spawner's built-in
  acquireRelease finalizer handles SIGTERM on scope close — no manual
  wiring needed.

- `handle.stdout` / `handle.stderr` are already Effect Streams, so the
  `fromBunStream` helper is gone (Stream.fromReadableStream + the
  per-pipe error-tag boilerplate it wrapped).

- acp's stdin moves from imperative `proc.stdin.write` + `proc.stdin.end`
  to a Queue<Uint8Array> fed into the spawner's stdin Sink via
  Stream.fromQueue. `send` is `Queue.offer`, `close` is `Queue.shutdown` —
  shutdown propagates as stdin EOF, which is ACP's graceful-exit signal.

- ServeHandle/AcpHandle public shape: `kill`/`close` become
  Effect<void> and `exited` becomes Effect<number> (was () => void and
  Promise<number>). The platform error that cross-spawn-spawner raises
  on signal-kill is collapsed to exit code -1 so `exited` stays a clean
  Effect<number> — matches the test contract (just needs proof of exit).

Two consuming tests updated to yield the Effect instead of awaiting
the Promise.

packages/opencode/test/cli/acp/acp-process.test.ts

@@ -51,19 +51,21 @@ describe("opencode acp (subprocess)", () => {
     "exits cleanly when stdin is closed (scope close)",
     ({ opencode }) =>
       Effect.gen(function* () {
-        const exitedPromise = yield* Effect.scoped(
+        const exited = yield* Effect.scoped(
           Effect.gen(function* () {
             const acp = yield* opencode.acp()
-            // Capture the Promise — scope-close fires the finalizer which
-            // ends stdin, and ACP should exit gracefully.
+            // Capture the Effect — scope-close shuts down stdinQueue, which
+            // propagates as stdin EOF; ACP exits gracefully. The exitCode
+            // Effect itself has no Scope requirement so yielding it after
+            // scope close is safe.
             return acp.exited
           }),
         )
 
-        const code = yield* Effect.promise(() => exitedPromise)
-        // Bun returns a number for normal exit. Anything goes for SIGTERM,
-        // but we still require resolution within the test timeout.
-        expect(typeof code === "number" || code === null).toBe(true)
+        const code = yield* exited
+        // Signal-killed processes surface as -1; clean EOF gives 0. Either
+        // way we just need a number — proves the process exited.
+        expect(typeof code).toBe("number")
       }),
     60_000,
   )

packages/opencode/test/cli/serve/serve-process.test.ts

@@ -41,20 +41,20 @@ describe("opencode serve (subprocess)", () => {
     ({ opencode }) =>
       Effect.gen(function* () {
         // Inner scope so we can observe `.exited` resolving after it closes.
-        const exitedPromise = yield* Effect.scoped(
+        const exited = yield* Effect.scoped(
           Effect.gen(function* () {
             const server = yield* opencode.serve()
-            // Capture the Promise, not the resolved value — scope closes after
-            // this gen returns, at which point the finalizer kills the child.
+            // Capture the Effect, not its result — scope closes after this
+            // gen returns, at which point the finalizer kills the child.
+            // handle.exitCode itself has no Scope requirement, so yielding
+            // it after scope close is fine.
             return server.exited
           }),
         )
         // After scope close: finalizer fired, process must have exited.
-        const code = yield* Effect.promise(() => exitedPromise)
-        // Bun reports the exit code; SIGTERM-killed processes return non-null
-        // (typically 143 on POSIX). We just require resolution within a sane
-        // window — anything else means the kill didn't take.
-        expect(typeof code === "number" || code === null).toBe(true)
+        // Signal-killed processes surface as -1 (see ServeHandle.exited).
+        const code = yield* exited
+        expect(typeof code).toBe("number")
       }),
     60_000,
   )

packages/opencode/test/lib/cli-process.ts

@@ -33,23 +33,13 @@ const cliEntry = path.join(opencodeRoot, "src/index.ts")
 
 export const testModelID = "test/test-model"
 
-// Wrap a Bun subprocess pipe (or any ReadableStream<Uint8Array>) as a Stream.
-// Centralizes the `evaluate` + `onError` boilerplate and tags errors with the
-// stream name so a stderr/stdout failure is greppable in logs.
-function fromBunStream(name: string, get: () => ReadableStream<Uint8Array>) {
-  return Stream.fromReadableStream({
-    evaluate: get,
-    onError: (cause) => new Error(`${name} stream error: ${String(cause)}`),
-  })
-}
-
 // Long-lived processes (serve, acp) all want the same stderr drain: read every
 // chunk, push to a tail buffer, swallow stream errors (the child closing the
 // pipe is normal). `log: true` surfaces a real protocol error to logs so a
 // regression doesn't silently disappear.
-function forkStderrDrain(stream: ReadableStream<Uint8Array>, into: string[]) {
+function forkStderrDrain(stream: Stream.Stream<Uint8Array, unknown>, into: string[]) {
   return Effect.forkScoped(
-    fromBunStream("stderr", () => stream).pipe(
+    stream.pipe(
       Stream.decodeText(),
       Stream.runForEach((chunk) => Effect.sync(() => into.push(chunk))),
       Effect.ignore({ log: true }),
@@ -117,9 +107,10 @@ export type ServeHandle = {
   readonly port: number
   // Sends SIGTERM. The scope finalizer also calls this, so tests rarely need
   // to invoke it directly — useful for tests that assert exit behavior.
-  readonly kill: () => void
-  // Resolves with the exit code once the process exits. Bun returns a number.
-  readonly exited: Promise<number>
+  readonly kill: Effect.Effect<void>
+  // Resolves with the exit code once the process exits. Signal-killed
+  // processes surface as -1 (vs cross-spawn-spawner raising a PlatformError).
+  readonly exited: Effect.Effect<number>
 }
 
 // `opencode acp` speaks newline-delimited JSON-RPC over stdin/stdout. It is
@@ -140,8 +131,10 @@ export type AcpHandle = {
   readonly receive: Effect.Effect<unknown>
   // Closes stdin. ACP exits cleanly on stdin EOF; the scope finalizer also
   // calls this, so tests only need it when asserting exit behavior.
-  readonly close: () => void
-  readonly exited: Promise<number>
+  readonly close: Effect.Effect<void>
+  // Resolves with the exit code once the process exits. Signal-killed
+  // processes surface as -1 (see ServeHandle.exited for the same convention).
+  readonly exited: Effect.Effect<number>
 }
 
 export type OpencodeCli = {
@@ -256,29 +249,22 @@ export function withCliFixture<A, E>(
       if (opts?.hostname) argv.push("--hostname", opts.hostname)
       if (opts?.extraArgs) argv.push(...opts.extraArgs)
 
-      // Acquire the subprocess; release sends SIGTERM and awaits exit on
-      // scope close. Wrapped in Effect.ignore so a flaky kill doesn't surface
-      // as a finalizer error during test teardown.
-      const proc = yield* Effect.acquireRelease(
-        Effect.sync(() =>
-          Bun.spawn(["bun", "run", "--conditions=browser", cliEntry, ...argv], {
-            cwd: home,
-            env: { ...process.env, ...env, ...opts?.env },
-            stdout: "pipe",
-            stderr: "pipe",
-          }),
-        ),
-        (p) =>
-          Effect.promise(() => {
-            p.kill()
-            return p.exited
-          }).pipe(Effect.ignore),
+      // ChildProcessSpawner.spawn returns a scoped handle whose acquireRelease
+      // finalizer sends SIGTERM and awaits exit on scope close — same lifecycle
+      // the old Bun.spawn + manual acquireRelease wrapper gave us, no plumbing.
+      const handle = yield* appProc.spawn(
+        ChildProcess.make("bun", ["run", "--conditions=browser", cliEntry, ...argv], {
+          cwd: home,
+          env: { ...env, ...opts?.env },
+          extendEnv: true,
+          stdin: "ignore",
+        }),
       )
 
       // Tail buffer so timeout failures can include stderr context. The fork
       // also keeps the OS pipe buffer from filling and wedging the child.
       const stderrChunks: string[] = []
-      yield* forkStderrDrain(proc.stderr, stderrChunks)
+      yield* forkStderrDrain(handle.stderr, stderrChunks)
 
       // Watch stdout line-by-line for the listening sentinel. Format
       // (see src/cli/cmd/serve.ts):
@@ -286,7 +272,7 @@ export function withCliFixture<A, E>(
       const readyRe = /listening on (http:\/\/([^\s:]+):(\d+))/
       const readyDeferred = yield* Deferred.make<{ url: string; hostname: string; port: number }>()
       yield* Effect.forkScoped(
-        fromBunStream("stdout", () => proc.stdout).pipe(
+        handle.stdout.pipe(
           Stream.decodeText(),
           Stream.splitLines,
           Stream.runForEach((line) => {
@@ -315,10 +301,14 @@ export function withCliFixture<A, E>(
         url: match.url,
         hostname: match.hostname,
         port: match.port,
-        kill: () => {
-          proc.kill()
-        },
-        exited: proc.exited as Promise<number>,
+        kill: handle.kill().pipe(Effect.ignore),
+        // handle.exitCode fails with PlatformError if the process was killed
+        // by signal (the normal scope-close path). Swallow → -1 so the test
+        // can still distinguish exited-vs-not without crashing.
+        exited: handle.exitCode.pipe(
+          Effect.orElseSucceed(() => -1),
+          Effect.map((c) => Number(c)),
+        ),
       } satisfies ServeHandle
     })
 
@@ -327,47 +317,30 @@ export function withCliFixture<A, E>(
       if (opts?.cwd) argv.push("--cwd", opts.cwd)
       if (opts?.extraArgs) argv.push(...opts.extraArgs)
 
-      // Acquire the subprocess. Release ends stdin (clean shutdown — ACP exits
-      // on stdin EOF) and falls back to SIGTERM if it doesn't exit promptly.
-      // Either way we await proc.exited so the test scope doesn't leak.
-      const proc = yield* Effect.acquireRelease(
-        Effect.sync(() =>
-          Bun.spawn(["bun", "run", "--conditions=browser", cliEntry, ...argv], {
-            cwd: opts?.cwd ?? home,
-            env: { ...process.env, ...env, ...opts?.env },
-            stdin: "pipe",
-            stdout: "pipe",
-            stderr: "pipe",
-          }),
-        ),
-        (p) =>
-          // Graceful shutdown: close stdin (ACP exits on EOF), give it a
-          // window to exit, then SIGTERM. The Effect.timeoutOrElse expresses
-          // exactly that race without raw setTimeout or Promise.race.
-          Effect.gen(function* () {
-            yield* Effect.sync(() => p.stdin.end())
-            yield* Effect.promise(() => p.exited).pipe(
-              Effect.timeoutOrElse({
-                duration: Duration.seconds(2),
-                orElse: () =>
-                  Effect.sync(() => {
-                    p.kill()
-                  }),
-              }),
-            )
-            yield* Effect.promise(() => p.exited)
-          }).pipe(Effect.ignore),
+      // stdin is fed by a Queue<Uint8Array>: send() offers bytes, close()
+      // shuts the queue down. The spawner drains the Queue-backed Stream into
+      // the child's stdin Sink (endOnDone: true by default), so a queue
+      // shutdown propagates as stdin EOF → ACP exits gracefully. Scope-close
+      // is the backstop via the spawner's kill finalizer.
+      const stdinQueue = yield* Queue.unbounded<Uint8Array>()
+      const handle = yield* appProc.spawn(
+        ChildProcess.make("bun", ["run", "--conditions=browser", cliEntry, ...argv], {
+          cwd: opts?.cwd ?? home,
+          env: { ...env, ...opts?.env },
+          extendEnv: true,
+          stdin: Stream.fromQueue(stdinQueue),
+        }),
       )
 
       const stderrChunks: string[] = []
-      yield* forkStderrDrain(proc.stderr, stderrChunks)
+      yield* forkStderrDrain(handle.stderr, stderrChunks)
 
       // Each ndjson line becomes one queue entry. JSON.parse failures are
       // surfaced as the raw string so a malformed protocol message doesn't
       // silently wedge the test in `receive`.
       const responses = yield* Queue.unbounded<unknown>()
       yield* Effect.forkScoped(
-        fromBunStream("stdout", () => proc.stdout).pipe(
+        handle.stdout.pipe(
           Stream.decodeText(),
           Stream.splitLines,
           Stream.runForEach((line) => {
@@ -384,20 +357,20 @@ export function withCliFixture<A, E>(
         ),
       )
 
+      const encoder = new TextEncoder()
       return {
-        // `proc.stdin.write` returns `number | Promise<number>`. The promise
-        // form is the backpressure signal — if we don't await it, rapid
-        // successive sends can interleave under pipe-buffer-full conditions
-        // and corrupt the ndjson framing.
         send: (msg: object) =>
-          Effect.promise(async () => {
-            const ret = proc.stdin.write(JSON.stringify(msg) + "\n")
-            if (typeof ret !== "number") await ret
-          }),
+          Queue.offer(stdinQueue, encoder.encode(JSON.stringify(msg) + "\n")).pipe(Effect.asVoid),
         receive: Queue.take(responses),
-        // proc.stdin.end() is idempotent in Bun; no try/catch needed.
-        close: () => proc.stdin.end(),
-        exited: proc.exited as Promise<number>,
+        // Queue shutdown → Stream.fromQueue completes → spawner ends stdin.
+        // Idempotent: shutting down an already-shut-down queue is a no-op.
+        close: Queue.shutdown(stdinQueue).pipe(Effect.asVoid),
+        // handle.exitCode fails with PlatformError on signal-kill; collapse
+        // to -1 to match the ServeHandle.exited convention.
+        exited: handle.exitCode.pipe(
+          Effect.orElseSucceed(() => -1),
+          Effect.map((c) => Number(c)),
+        ),
       } satisfies AcpHandle
     })