refactor(acp): make the ACP request boundary effect-first

packages/cli/src/acp/agent.ts

@@ -1,4 +1,13 @@
-import { agent, RequestError, type Stream } from "@agentclientprotocol/sdk"
+import {
+  agent,
+  RequestError,
+  type AgentHandlerContext,
+  type AgentNotificationHandlersByMethod,
+  type AgentNotificationMethod,
+  type AgentRequestHandlersByMethod,
+  type AgentRequestMethod,
+  type Stream,
+} from "@agentclientprotocol/sdk"
 import { ClientError, type OpenCodeClient } from "@opencode/client/promise"
 import { Cause, Effect } from "effect"
 import { ACPConnection } from "./connection"
@@ -8,83 +17,97 @@ import { ACPService } from "./service"
 // 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 handle =
+    <Params, A>(call: (ctx: AgentHandlerContext<Params>) => Effect.Effect<A, ACPError.Error | RequestError>) =>
+    (name: string) => {
+      const handler = Effect.fn(name)(
+        call,
+        Effect.mapError((error) => (error instanceof RequestError ? error : ACPError.toRequestError(error))),
+        Effect.tapCauseIf(Cause.hasDies, (cause) => Effect.logError("ACP request failed", cause)),
+        Effect.catchDefect((defect) => Effect.fail(ACPError.toRequestError(ACPError.fromUnknown(defect)))),
+      )
+      return (ctx: AgentHandlerContext<Params>) => run(handler(ctx))
     }
-  const connection = agent({ name: "opencode" })
-    .onRequest(
+  const app = agent({ name: "opencode" })
+  const request = <Method extends AgentRequestMethod>(
+    method: Method,
+    make: (name: string) => AgentRequestHandlersByMethod[Method],
+  ) => app.onRequest(method, make(spanName(method)))
+  const notification = <Method extends AgentNotificationMethod>(
+    method: Method,
+    make: (name: string) => AgentNotificationHandlersByMethod[Method],
+  ) => app.onNotification(method, make(spanName(method)))
+
+  request(
     "initialize",
-      handle("ACP.initialize", (params) => service.initialize(params)),
+    handle((ctx) => promise(() => service.initialize(ctx.params))),
   )
-    .onRequest(
+  request(
     "authenticate",
-      handle("ACP.authenticate", (params) => service.authenticate(params)),
+    handle((ctx) => promise(() => service.authenticate(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/new",
-      handle("ACP.session.new", (params) => service.newSession(params)),
+    handle((ctx) => promise(() => service.newSession(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/load",
-      handle("ACP.session.load", (params) => service.loadSession(params)),
+    handle((ctx) => promise(() => service.loadSession(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/list",
-      handle("ACP.session.list", (params) => service.listSessions(params)),
+    handle((ctx) => promise(() => service.listSessions(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/delete",
-      handle("ACP.session.delete", (params) => service.deleteSession(params)),
+    handle((ctx) => promise(() => service.deleteSession(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/resume",
-      handle("ACP.session.resume", (params) => service.resumeSession(params)),
+    handle((ctx) => promise(() => service.resumeSession(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/close",
-      handle("ACP.session.close", (params) => service.closeSession(params)),
+    handle((ctx) => promise(() => service.closeSession(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/fork",
-      handle("ACP.session.fork", (params) => service.forkSession(params)),
+    handle((ctx) => promise(() => service.forkSession(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/set_config_option",
-      handle("ACP.session.set_config_option", (params) => service.setSessionConfigOption(params)),
+    handle((ctx) => promise(() => service.setSessionConfigOption(ctx.params))),
   )
-    .onRequest(
+  request(
     "session/set_mode",
-      handle("ACP.session.set_mode", (params) => service.setSessionMode(params)),
+    handle((ctx) => promise(() => service.setSessionMode(ctx.params))),
   )
   // The SDK signal is passed through rather than interrupting the fiber: a cancelled turn still resolves with
   // `stopReason: "cancelled"`.
-    .onRequest(
+  request(
     "session/prompt",
-      handle("ACP.session.prompt", (params, signal) => service.prompt(params, signal)),
+    handle((ctx) => promise(() => service.prompt(ctx.params, ctx.signal))),
   )
-    .onNotification(
+  notification(
     "session/cancel",
-      handle("ACP.session.cancel", (params) => service.cancel(params)),
+    handle((ctx) => promise(() => service.cancel(ctx.params))),
   )
-    .connect(stream)
+  const connection = app.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
 })
 
-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)))
-})
+const spanName = (method: string) => `cli.acp.${method.replaceAll("/", ".")}`
+
+const promise = <A>(evaluate: () => Promise<A>) =>
+  Effect.tryPromise({ try: evaluate, catch: (cause) => cause }).pipe(
+    Effect.catch((cause) => {
+      if (cause instanceof RequestError || ACPError.is(cause)) return Effect.fail(cause)
+      if (cause instanceof ClientError && cause.reason === "Transport")
+        return Effect.fail(new ACPError.ServerUnavailableError())
+      return Effect.die(cause)
+    }),
+  )
 
 export * as ACP from "./agent"

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

@@ -1,7 +1,9 @@
 import { ndJsonStream } from "@agentclientprotocol/sdk"
 import { OpenCode } from "@opencode/client/promise"
 import { Service } from "@opencode/client/effect/service"
+import { CrossSpawnSpawner } from "@opencode/util/cross-spawn-spawner"
 import { Effect } from "effect"
+import { Writable } from "node:stream"
 import { ACP } from "../../acp/agent"
 import { Commands } from "../commands"
 import { Runtime } from "../../framework/runtime"
@@ -13,43 +15,26 @@ 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 output = new WritableStream<Uint8Array>({
-      write: (chunk) =>
-        new Promise<void>((resolve, reject) => {
-          process.stdout.write(chunk, (error) => (error ? reject(error) : resolve()))
+    const connection = yield* ACP.connect(client, ndJsonStream(Writable.toWeb(process.stdout), Bun.stdin.stream()))
+    const code = yield* Effect.raceFirst(
+      Effect.promise(() => connection.closed).pipe(Effect.as(0)),
+      endpoint.exited.pipe(
+        Effect.match({
+          onSuccess: (code) => `code ${code}`,
+          onFailure: (error) =>
+            error.cause instanceof CrossSpawnSpawner.KilledBySignal ? `signal ${error.cause.signal}` : error.message,
         }),
-    })
-    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.
-    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(() => {
+        Effect.flatMap((reason) =>
+          Effect.sync(() => {
             process.stderr.write(`opencode acp: server exited unexpectedly (${reason})\n`)
-      process.exit(1)
-    })
+            return 1
+          }),
+        ),
+      ),
+    )
+    // Closing the handler scope would wait for the private server's graceful shutdown; its lease pipe already
+    // ends the server once this process exits.
+    yield* Effect.sync(() => process.exit(code))
   }),
 )
-
-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

@@ -1,4 +1,4 @@
-import { Service, type Endpoint } from "@opencode/client/effect/service"
+import { Service } from "@opencode/client/effect/service"
 import { CrossSpawnSpawner } from "@opencode/util/cross-spawn-spawner"
 import { LayerNode } from "@opencode/util/effect/layer-node"
 import { Deferred, Effect, Schema, Stream } from "effect"
@@ -13,8 +13,6 @@ 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) {
@@ -55,15 +53,8 @@ const makeEndpoint = Effect.fn("cli.standalone.endpoint")(
       url: ready.url,
       auth: { type: "basic" as const, username: "opencode", password },
       pid: proc.pid,
-      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> }
+      exited: proc.exitCode,
+    }
   },
   Effect.provide(LayerNode.compile(CrossSpawnSpawner.node)),
 )

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

@@ -1,5 +1,6 @@
 import { describe, expect, test } from "bun:test"
 import { RequestError } from "@agentclientprotocol/sdk"
+import { Cause } from "effect"
 import { ACPError } from "../../src/acp/error"
 import { rpcError, startSession, startWire } from "./wire-fixture"
 
@@ -73,6 +74,9 @@ describe("acp error boundary over the wire", () => {
       message: "Internal error: Internal service failure",
       data: { errorName: "ClientError" },
     })
+    expect(acp.logs.map((log) => ({ message: log.message, cause: Cause.squash(log.cause) }))).toMatchObject([
+      { message: ["ACP request failed"], cause: { name: "ClientError", reason: "UnexpectedStatus" } },
+    ])
   })
 
   test("reports an unavailable server once the server stops", async () => {
@@ -84,5 +88,6 @@ describe("acp error boundary over the wire", () => {
       message: "Internal error: OpenCode server is unavailable",
       data: { errorName: "ServerUnavailable" },
     })
+    expect(acp.logs).toEqual([])
   })
 })

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

@@ -28,7 +28,7 @@ import {
   type TokenUsageInfo,
 } from "@opencode/client/promise"
 import type { BunRequest } from "bun"
-import { Effect, Option, Schema } from "effect"
+import { Effect, Logger, Option, Schema } from "effect"
 import { ACP } from "../../src/acp/agent"
 
 type DurableEvent = Extract<OpenCodeEvent, { durable: unknown }>
@@ -377,8 +377,12 @@ export async function startWire(options: WireOptions = {}) {
 
   const clientToAgent = new TransformStream<Uint8Array, Uint8Array>()
   const agentToClient = new TransformStream<Uint8Array, Uint8Array>()
+  const logs: Array<Pick<Logger.Options<unknown>, "message" | "cause">> = []
   const agentConnection = await Effect.runPromise(
-    ACP.connect(OpenCode.make({ baseUrl: server.url }), ndJsonStream(agentToClient.writable, clientToAgent.readable)),
+    ACP.connect(
+      OpenCode.make({ baseUrl: server.url }),
+      ndJsonStream(agentToClient.writable, clientToAgent.readable),
+    ).pipe(Effect.provide(Logger.layer([Logger.make((log) => logs.push({ message: log.message, cause: log.cause }))]))),
   )
   const clientStream = ndJsonStream(clientToAgent.writable, agentToClient.readable)
   const connection = client({ name: "test" })
@@ -455,6 +459,7 @@ export async function startWire(options: WireOptions = {}) {
 
   return {
     server,
+    logs,
     received,
     updates,
     permissions,