From 138e5f26b2657c274684a125e9e3547dfc4abe65 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Thu, 1 Oct 2026 00:55:45 +0530 Subject: [PATCH] refactor(acp): make the ACP request boundary effect-first --- packages/cli/src/acp/agent.ts | 171 ++++++++++++---------- packages/cli/src/commands/handlers/acp.ts | 57 +++----- packages/cli/src/services/standalone.ts | 15 +- packages/cli/test/acp/error.test.ts | 5 + packages/cli/test/acp/wire-fixture.ts | 9 +- 5 files changed, 133 insertions(+), 124 deletions(-) diff --git a/packages/cli/src/acp/agent.ts b/packages/cli/src/acp/agent.ts index ece015a0843..40ab20747e6 100644 --- a/packages/cli/src/acp/agent.ts +++ b/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()) - const handle = (name: string, call: (params: Params, signal: AbortSignal) => Promise) => { - 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", - 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) + const handle = + (call: (ctx: AgentHandlerContext) => Effect.Effect) => + (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) => run(handler(ctx)) + } + const app = agent({ name: "opencode" }) + const request = ( + method: Method, + make: (name: string) => AgentRequestHandlersByMethod[Method], + ) => app.onRequest(method, make(spanName(method))) + const notification = ( + method: Method, + make: (name: string) => AgentNotificationHandlersByMethod[Method], + ) => app.onNotification(method, make(spanName(method))) + + request( + "initialize", + handle((ctx) => promise(() => service.initialize(ctx.params))), + ) + request( + "authenticate", + handle((ctx) => promise(() => service.authenticate(ctx.params))), + ) + request( + "session/new", + handle((ctx) => promise(() => service.newSession(ctx.params))), + ) + request( + "session/load", + handle((ctx) => promise(() => service.loadSession(ctx.params))), + ) + request( + "session/list", + handle((ctx) => promise(() => service.listSessions(ctx.params))), + ) + request( + "session/delete", + handle((ctx) => promise(() => service.deleteSession(ctx.params))), + ) + request( + "session/resume", + handle((ctx) => promise(() => service.resumeSession(ctx.params))), + ) + request( + "session/close", + handle((ctx) => promise(() => service.closeSession(ctx.params))), + ) + request( + "session/fork", + handle((ctx) => promise(() => service.forkSession(ctx.params))), + ) + request( + "session/set_config_option", + handle((ctx) => promise(() => service.setSessionConfigOption(ctx.params))), + ) + request( + "session/set_mode", + 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"`. + request( + "session/prompt", + handle((ctx) => promise(() => service.prompt(ctx.params, ctx.signal))), + ) + notification( + "session/cancel", + handle((ctx) => promise(() => service.cancel(ctx.params))), + ) + 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) { - 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 = (evaluate: () => Promise) => + 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" diff --git a/packages/cli/src/commands/handlers/acp.ts b/packages/cli/src/commands/handlers/acp.ts index 37b29cee82e..56ece4f82cf 100644 --- a/packages/cli/src/commands/handlers/acp.ts +++ b/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({ - write: (chunk) => - new Promise((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, + // stdout carries ACP, so the diagnostic goes to stderr. + Effect.flatMap((reason) => + Effect.sync(() => { + process.stderr.write(`opencode acp: server exited unexpectedly (${reason})\n`) + return 1 + }), + ), + ), ) - // 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(() => { - process.stderr.write(`opencode acp: server exited unexpectedly (${reason})\n`) - process.exit(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 } = {} - 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({ 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() - }), - ) -}) diff --git a/packages/cli/src/services/standalone.ts b/packages/cli/src/services/standalone.ts index 8054307c483..b69e13ea3ea 100644 --- a/packages/cli/src/services/standalone.ts +++ b/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 } -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({ signal: error.cause.signal }) - : Effect.die(error), - ), - ), - } satisfies Endpoint & { readonly pid: number; readonly exited: Effect.Effect } + exited: proc.exitCode, + } }, Effect.provide(LayerNode.compile(CrossSpawnSpawner.node)), ) diff --git a/packages/cli/test/acp/error.test.ts b/packages/cli/test/acp/error.test.ts index 101b2da8f91..b17176764d5 100644 --- a/packages/cli/test/acp/error.test.ts +++ b/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([]) }) }) diff --git a/packages/cli/test/acp/wire-fixture.ts b/packages/cli/test/acp/wire-fixture.ts index b978452f12f..fbe5cff5627 100644 --- a/packages/cli/test/acp/wire-fixture.ts +++ b/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 @@ -377,8 +377,12 @@ export async function startWire(options: WireOptions = {}) { const clientToAgent = new TransformStream() const agentToClient = new TransformStream() + const logs: Array, "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,