refactor(acp): run ACP requests as effects and exit when the server dies

packages/cli/src/acp/agent.ts

@@ -1,52 +1,90 @@
 import { agent, RequestError, type Stream } from "@agentclientprotocol/sdk"
-import type { OpenCodeClient } from "@opencode/client/promise"
+import { ClientError, type OpenCodeClient } from "@opencode/client/promise"
+import { Cause, Effect } from "effect"
 import { ACPConnection } from "./connection"
 import { ACPError } from "./error"
 import { ACPService } from "./service"
 
-export function connect(client: OpenCodeClient, stream: Stream) {
+// Untraced so request spans parent to the caller's span instead of a setup span that has already ended.
+export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stream: Stream) {
+  const run = Effect.runPromiseWith(yield* Effect.context<never>())
+  const handle = <Params, A>(name: string, call: (params: Params, signal: AbortSignal) => Promise<A>) => {
+    const handler = Effect.fn(name)(function* (params: Params, signal: AbortSignal) {
+      return yield* Effect.tryPromise({
+        try: () => call(params, signal),
+        catch: (cause) =>
+          cause instanceof ClientError && cause.reason === "Transport" ? new ACPError.ServerUnavailableError() : cause,
+      })
+    }, Effect.catchCause(toRequestError))
+    return (ctx: { readonly params: Params; readonly signal: AbortSignal }) => run(handler(ctx.params, ctx.signal))
+  }
   const connection = agent({ name: "opencode" })
-    .onRequest("initialize", (ctx) => run(service.initialize(ctx.params)))
-    .onRequest("authenticate", (ctx) => run(service.authenticate(ctx.params)))
-    .onRequest("session/new", (ctx) => run(service.newSession(ctx.params)))
-    .onRequest("session/load", (ctx) => run(service.loadSession(ctx.params)))
-    .onRequest("session/list", (ctx) => run(service.listSessions(ctx.params)))
-    .onRequest("session/delete", (ctx) => run(service.deleteSession(ctx.params)))
-    .onRequest("session/resume", (ctx) => run(service.resumeSession(ctx.params)))
-    .onRequest("session/close", (ctx) => run(service.closeSession(ctx.params)))
-    .onRequest("session/fork", (ctx) => run(service.forkSession(ctx.params)))
-    .onRequest("session/set_config_option", (ctx) => run(service.setSessionConfigOption(ctx.params)))
-    .onRequest("session/set_mode", (ctx) => run(service.setSessionMode(ctx.params)))
-    .onRequest("session/prompt", (ctx) => run(service.prompt(ctx.params, ctx.signal)))
-    .onNotification("session/cancel", (ctx) => run(service.cancel(ctx.params)))
+    .onRequest(
+      "initialize",
+      handle("ACP.initialize", (params) => service.initialize(params)),
+    )
+    .onRequest(
+      "authenticate",
+      handle("ACP.authenticate", (params) => service.authenticate(params)),
+    )
+    .onRequest(
+      "session/new",
+      handle("ACP.session.new", (params) => service.newSession(params)),
+    )
+    .onRequest(
+      "session/load",
+      handle("ACP.session.load", (params) => service.loadSession(params)),
+    )
+    .onRequest(
+      "session/list",
+      handle("ACP.session.list", (params) => service.listSessions(params)),
+    )
+    .onRequest(
+      "session/delete",
+      handle("ACP.session.delete", (params) => service.deleteSession(params)),
+    )
+    .onRequest(
+      "session/resume",
+      handle("ACP.session.resume", (params) => service.resumeSession(params)),
+    )
+    .onRequest(
+      "session/close",
+      handle("ACP.session.close", (params) => service.closeSession(params)),
+    )
+    .onRequest(
+      "session/fork",
+      handle("ACP.session.fork", (params) => service.forkSession(params)),
+    )
+    .onRequest(
+      "session/set_config_option",
+      handle("ACP.session.set_config_option", (params) => service.setSessionConfigOption(params)),
+    )
+    .onRequest(
+      "session/set_mode",
+      handle("ACP.session.set_mode", (params) => service.setSessionMode(params)),
+    )
+    // The SDK signal is passed through rather than interrupting the fiber: a cancelled turn still resolves with
+    // `stopReason: "cancelled"`.
+    .onRequest(
+      "session/prompt",
+      handle("ACP.session.prompt", (params, signal) => service.prompt(params, signal)),
+    )
+    .onNotification(
+      "session/cancel",
+      handle("ACP.session.cancel", (params) => service.cancel(params)),
+    )
     .connect(stream)
   // Inbound dispatch starts after the stream's async read loop yields, so handlers never observe this before assignment.
   const service = ACPService.make({ client, connection: ACPConnection.make(connection) })
   return connection
-}
+})
 
-async function run<A>(promise: Promise<A>) {
-  try {
-    return await promise
-  } catch (error) {
-    if (error instanceof RequestError) throw error
-    if (isACPError(error)) throw ACPError.toRequestError(error)
-    throw ACPError.toRequestError(ACPError.fromUnknown(error))
-  }
-}
-
-function isACPError(error: unknown): error is ACPError.Error {
-  return (
-    error instanceof ACPError.SessionNotFoundError ||
-    error instanceof ACPError.SessionDirectoryMismatchError ||
-    error instanceof ACPError.InvalidConfigOptionError ||
-    error instanceof ACPError.InvalidModelError ||
-    error instanceof ACPError.InvalidEffortError ||
-    error instanceof ACPError.InvalidModeError ||
-    error instanceof ACPError.AuthRequiredError ||
-    error instanceof ACPError.UnknownAuthMethodError ||
-    error instanceof ACPError.ServiceFailureError
-  )
-}
+const toRequestError = Effect.fnUntraced(function* (cause: Cause.Cause<unknown>) {
+  const error = Cause.squash(cause)
+  if (error instanceof RequestError) return yield* Effect.fail(error)
+  if (ACPError.is(error)) return yield* Effect.fail(ACPError.toRequestError(error))
+  yield* Effect.logError("ACP request failed", cause)
+  return yield* Effect.fail(ACPError.toRequestError(ACPError.fromUnknown(error)))
+})
 
 export * as ACP from "./agent"

packages/cli/src/acp/error.ts

@@ -40,16 +40,27 @@ export class ServiceFailureError extends Schema.TaggedError<ServiceFailureError>
   errorName: Schema.optional(Schema.String),
 }) {}
 
-export type Error =
-  | SessionNotFoundError
-  | SessionDirectoryMismatchError
-  | InvalidConfigOptionError
-  | InvalidModelError
-  | InvalidEffortError
-  | InvalidModeError
-  | AuthRequiredError
-  | UnknownAuthMethodError
-  | ServiceFailureError
+export class ServerUnavailableError extends Schema.TaggedError<ServerUnavailableError>()(
+  "ACPServerUnavailableError",
+  {},
+) {}
+
+const Errors = Schema.Union([
+  SessionNotFoundError,
+  SessionDirectoryMismatchError,
+  InvalidConfigOptionError,
+  InvalidModelError,
+  InvalidEffortError,
+  InvalidModeError,
+  AuthRequiredError,
+  UnknownAuthMethodError,
+  ServiceFailureError,
+  ServerUnavailableError,
+])
+
+export type Error = typeof Errors.Type
+
+export const is = Schema.is(Errors)
 
 export function toRequestError(error: Error): RequestError {
   switch (error._tag) {
@@ -83,6 +94,8 @@ export function toRequestError(error: Error): RequestError {
         },
         error.safeMessage,
       )
+    case "ACPServerUnavailableError":
+      return RequestError.internalError({ errorName: "ServerUnavailable" }, "OpenCode server is unavailable")
   }
   const exhaustive: never = error
   return exhaustive

packages/cli/src/commands/handlers/acp.ts

@@ -13,23 +13,43 @@ export default Runtime.handler(
     process.env.OPENCODE_CLIENT = "acp"
     const endpoint = yield* Standalone.start()
     const client = OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint) })
-    const input = new WritableStream<Uint8Array>({
+    const output = new WritableStream<Uint8Array>({
       write: (chunk) =>
         new Promise<void>((resolve, reject) => {
           process.stdout.write(chunk, (error) => (error ? reject(error) : resolve()))
         }),
     })
-    const output = new ReadableStream<Uint8Array>({
-      start(controller) {
-        process.stdin.on("data", (chunk: Buffer) => controller.enqueue(new Uint8Array(chunk)))
-        process.stdin.on("end", () => controller.close())
-        process.stdin.on("error", (error) => controller.error(error))
-      },
-    })
-    const connection = ACP.connect(client, ndJsonStream(input, output))
-    process.stdin.resume()
-    yield* Effect.promise(() => connection.closed)
+    const connection = yield* ACP.connect(client, ndJsonStream(output, yield* stdin))
+    const exited = yield* Effect.raceFirst(
+      Effect.promise(() => connection.closed).pipe(Effect.as(undefined)),
+      endpoint.exited,
+    )
     // EOF owns this stdio process; exiting also closes the private server's lease pipe.
-    yield* Effect.sync(() => process.exit(0))
+    if (!exited) return yield* Effect.sync(() => process.exit(0))
+    const reason = "code" in exited ? `code ${exited.code}` : `signal ${exited.signal}`
+    // stdout carries ACP, so the diagnostic goes to stderr.
+    yield* Effect.sync(() => {
+      process.stderr.write(`opencode acp: server exited unexpectedly (${reason})\n`)
+      process.exit(1)
+    })
   }),
 )
+
+const stdin = Effect.suspend(() => {
+  const source: { controller?: ReadableStreamDefaultController<Uint8Array> } = {}
+  const data = (chunk: Buffer) => source.controller?.enqueue(new Uint8Array(chunk))
+  const end = () => source.controller?.close()
+  const error = (cause: Error) => source.controller?.error(cause)
+  return Effect.acquireRelease(
+    Effect.sync(() => {
+      const stream = new ReadableStream<Uint8Array>({ start: (controller) => void (source.controller = controller) })
+      process.stdin.on("data", data).on("end", end).on("error", error)
+      return stream
+    }),
+    () =>
+      Effect.sync(() => {
+        process.stdin.off("data", data).off("end", end).off("error", error)
+        process.stdin.pause()
+      }),
+  )
+})

packages/cli/src/services/standalone.ts

@@ -13,6 +13,8 @@ type Options = {
   readonly command?: ReadonlyArray<string>
 }
 
+export type Exit = { readonly code: number } | { readonly signal: string }
+
 const startupDirectory = process.cwd()
 
 function command(password: string, options: Options) {
@@ -53,7 +55,15 @@ const makeEndpoint = Effect.fn("cli.standalone.endpoint")(
       url: ready.url,
       auth: { type: "basic" as const, username: "opencode", password },
       pid: proc.pid,
-    } satisfies Endpoint & { readonly pid: number }
+      exited: proc.exitCode.pipe(
+        Effect.map((code): Exit => ({ code })),
+        Effect.catch((error) =>
+          error.cause instanceof CrossSpawnSpawner.KilledBySignal
+            ? Effect.succeed<Exit>({ signal: error.cause.signal })
+            : Effect.die(error),
+        ),
+      ),
+    } satisfies Endpoint & { readonly pid: number; readonly exited: Effect.Effect<Exit> }
   },
   Effect.provide(LayerNode.compile(CrossSpawnSpawner.node)),
 )

packages/cli/test/acp/error.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, test } from "bun:test"
 import { RequestError } from "@agentclientprotocol/sdk"
 import { ACPError } from "../../src/acp/error"
+import { rpcError, startSession, startWire } from "./wire-fixture"
 
 describe("acp errors", () => {
   test("maps validation failures to invalid params", () => {
@@ -58,3 +59,30 @@ describe("acp errors", () => {
     expect(serialized).not.toContain("stack")
   })
 })
+
+describe("acp error boundary over the wire", () => {
+  test("maps unexpected server failures to the generic internal error", async () => {
+    await using acp = await startWire({
+      fetch: (request) =>
+        request.method === "POST" && request.path === "/api/session" ? new Response(null, { status: 500 }) : undefined,
+    })
+    await acp.initialize()
+
+    expect(await rpcError(acp.newSession())).toEqual({
+      code: -32603,
+      message: "Internal error: Internal service failure",
+      data: { errorName: "ClientError" },
+    })
+  })
+
+  test("reports an unavailable server once the server stops", async () => {
+    await using acp = await startSession()
+    await acp.server.stop()
+
+    expect(await rpcError(acp.request("session/list", {}))).toEqual({
+      code: -32603,
+      message: "Internal error: OpenCode server is unavailable",
+      data: { errorName: "ServerUnavailable" },
+    })
+  })
+})

packages/cli/test/acp/lifecycle.subprocess.test.ts

@@ -101,8 +101,8 @@ describe("acp lifecycle subprocess", () => {
   }, 60_000)
 
   // The private server is found with `pgrep`, which Windows lacks.
-  const todoOutsideWindows = process.platform === "win32" ? test.skip : test.todo
-  todoOutsideWindows(
+  const testOutsideWindows = process.platform === "win32" ? test.skip : test
+  testOutsideWindows(
     "exits when the private server process dies (https://github.com/anomalyco/opencode/issues/51716)",
     async () => {
       await using fixture = await createAcpFixture()
@@ -121,7 +121,9 @@ describe("acp lifecycle subprocess", () => {
       const timeout = Promise.withResolvers<"running">()
       const timer = setTimeout(() => timeout.resolve("running"), 10_000)
       const exited = await Promise.race([acp.exited, timeout.promise]).finally(() => clearTimeout(timer))
-      expect(exited).not.toBe("running")
+      expect(exited).toBe(1)
+      await acp[Symbol.asyncDispose]()
+      expect(acp.stderr()).toContain("opencode acp: server exited unexpectedly (signal SIGKILL)")
     },
     60_000,
   )

packages/cli/test/acp/wire-fixture.ts

@@ -28,7 +28,7 @@ import {
   type TokenUsageInfo,
 } from "@opencode/client/promise"
 import type { BunRequest } from "bun"
-import { Option, Schema } from "effect"
+import { Effect, Option, Schema } from "effect"
 import { ACP } from "../../src/acp/agent"
 
 type DurableEvent = Extract<OpenCodeEvent, { durable: unknown }>
@@ -377,9 +377,8 @@ export async function startWire(options: WireOptions = {}) {
 
   const clientToAgent = new TransformStream<Uint8Array, Uint8Array>()
   const agentToClient = new TransformStream<Uint8Array, Uint8Array>()
-  const agentConnection = ACP.connect(
-    OpenCode.make({ baseUrl: server.url }),
-    ndJsonStream(agentToClient.writable, clientToAgent.readable),
+  const agentConnection = await Effect.runPromise(
+    ACP.connect(OpenCode.make({ baseUrl: server.url }), ndJsonStream(agentToClient.writable, clientToAgent.readable)),
   )
   const clientStream = ndJsonStream(clientToAgent.writable, agentToClient.readable)
   const connection = client({ name: "test" })