diff --git a/packages/cli/src/acp/agent.ts b/packages/cli/src/acp/agent.ts index fba6525a451..ece015a0843 100644 --- a/packages/cli/src/acp/agent.ts +++ b/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()) + 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", (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(promise: Promise) { - 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) { + 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" diff --git a/packages/cli/src/acp/error.ts b/packages/cli/src/acp/error.ts index 6256ae92b44..6ffc0c88743 100644 --- a/packages/cli/src/acp/error.ts +++ b/packages/cli/src/acp/error.ts @@ -40,16 +40,27 @@ export class ServiceFailureError extends Schema.TaggedError errorName: Schema.optional(Schema.String), }) {} -export type Error = - | SessionNotFoundError - | SessionDirectoryMismatchError - | InvalidConfigOptionError - | InvalidModelError - | InvalidEffortError - | InvalidModeError - | AuthRequiredError - | UnknownAuthMethodError - | ServiceFailureError +export class ServerUnavailableError extends Schema.TaggedError()( + "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 diff --git a/packages/cli/src/commands/handlers/acp.ts b/packages/cli/src/commands/handlers/acp.ts index e170f05bc06..37b29cee82e 100644 --- a/packages/cli/src/commands/handlers/acp.ts +++ b/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({ + const output = new WritableStream({ write: (chunk) => new Promise((resolve, reject) => { process.stdout.write(chunk, (error) => (error ? reject(error) : resolve())) }), }) - const output = new ReadableStream({ - 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 } = {} + 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 9ec70ac8431..8054307c483 100644 --- a/packages/cli/src/services/standalone.ts +++ b/packages/cli/src/services/standalone.ts @@ -13,6 +13,8 @@ type Options = { readonly command?: ReadonlyArray } +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({ signal: error.cause.signal }) + : Effect.die(error), + ), + ), + } satisfies Endpoint & { readonly pid: number; readonly exited: Effect.Effect } }, 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 421364d0feb..101b2da8f91 100644 --- a/packages/cli/test/acp/error.test.ts +++ b/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" }, + }) + }) +}) diff --git a/packages/cli/test/acp/lifecycle.subprocess.test.ts b/packages/cli/test/acp/lifecycle.subprocess.test.ts index 72d2a210300..0e5c2f2886f 100644 --- a/packages/cli/test/acp/lifecycle.subprocess.test.ts +++ b/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, ) diff --git a/packages/cli/test/acp/wire-fixture.ts b/packages/cli/test/acp/wire-fixture.ts index 01e3da975c7..b978452f12f 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 { Option, Schema } from "effect" +import { Effect, Option, Schema } from "effect" import { ACP } from "../../src/acp/agent" type DurableEvent = Extract @@ -377,9 +377,8 @@ export async function startWire(options: WireOptions = {}) { const clientToAgent = new TransformStream() const agentToClient = new TransformStream() - 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" })