diff --git a/packages/cli/src/acp/agent.ts b/packages/cli/src/acp/agent.ts index 2d962c2b734..fba6525a451 100644 --- a/packages/cli/src/acp/agent.ts +++ b/packages/cli/src/acp/agent.ts @@ -1,42 +1,28 @@ -import { - RequestError, - type Agent, - type AgentSideConnection, - type AuthenticateRequest, - type CancelNotification, - type CloseSessionRequest, - type DeleteSessionRequest, - type ForkSessionRequest, - type InitializeRequest, - type ListSessionsRequest, - type LoadSessionRequest, - type NewSessionRequest, - type PromptRequest, - type ResumeSessionRequest, - type SetSessionConfigOptionRequest, - type SetSessionModeRequest, -} from "@agentclientprotocol/sdk" +import { agent, RequestError, type Stream } from "@agentclientprotocol/sdk" import type { OpenCodeClient } from "@opencode/client/promise" +import { ACPConnection } from "./connection" import { ACPError } from "./error" import { ACPService } from "./service" -export function create(client: OpenCodeClient, connection: AgentSideConnection) { - const service = ACPService.make({ client, connection }) - return { - initialize: (params: InitializeRequest) => run(service.initialize(params)), - authenticate: (params: AuthenticateRequest) => run(service.authenticate(params)), - newSession: (params: NewSessionRequest) => run(service.newSession(params)), - loadSession: (params: LoadSessionRequest) => run(service.loadSession(params)), - listSessions: (params: ListSessionsRequest) => run(service.listSessions(params)), - deleteSession: (params: DeleteSessionRequest) => run(service.deleteSession(params)), - resumeSession: (params: ResumeSessionRequest) => run(service.resumeSession(params)), - closeSession: (params: CloseSessionRequest) => run(service.closeSession(params)), - unstable_forkSession: (params: ForkSessionRequest) => run(service.forkSession(params)), - setSessionConfigOption: (params: SetSessionConfigOptionRequest) => run(service.setSessionConfigOption(params)), - setSessionMode: (params: SetSessionModeRequest) => run(service.setSessionMode(params)), - prompt: (params: PromptRequest) => run(service.prompt(params)), - cancel: (params: CancelNotification) => run(service.cancel(params)), - } satisfies Agent +export function connect(client: OpenCodeClient, stream: Stream) { + 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))) + .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) { @@ -52,6 +38,7 @@ async function run(promise: Promise) { 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 || diff --git a/packages/cli/src/acp/connection.ts b/packages/cli/src/acp/connection.ts new file mode 100644 index 00000000000..5b75771a44e --- /dev/null +++ b/packages/cli/src/acp/connection.ts @@ -0,0 +1,31 @@ +import { + methods, + type AgentConnection, + type RequestPermissionRequest, + type RequestPermissionResponse, + type SendRequestOptions, + type SessionNotification, + type WriteTextFileRequest, + type WriteTextFileResponse, +} from "@agentclientprotocol/sdk" + +export type Connection = { + readonly signal?: AbortSignal + sessionUpdate(params: SessionNotification): Promise + requestPermission(params: RequestPermissionRequest, options?: SendRequestOptions): Promise + writeTextFile?(params: WriteTextFileRequest, options?: SendRequestOptions): Promise + extNotification?(method: string, params: Record): Promise +} + +export function make(connection: AgentConnection): Connection { + return { + signal: connection.signal, + sessionUpdate: (params) => connection.client.notify(methods.client.session.update, params), + requestPermission: (params, options) => + connection.client.request(methods.client.session.requestPermission, params, options), + writeTextFile: (params, options) => connection.client.request(methods.client.fs.writeTextFile, params, options), + extNotification: (method, params) => connection.client.notify(method, params), + } +} + +export * as ACPConnection from "./connection" diff --git a/packages/cli/src/acp/error.ts b/packages/cli/src/acp/error.ts index fcdd3bc65c1..6256ae92b44 100644 --- a/packages/cli/src/acp/error.ts +++ b/packages/cli/src/acp/error.ts @@ -5,6 +5,11 @@ export class SessionNotFoundError extends Schema.TaggedError()( + "ACPSessionDirectoryMismatchError", + { sessionId: Schema.String, cwd: Schema.String }, +) {} + export class InvalidConfigOptionError extends Schema.TaggedError()( "ACPInvalidConfigOptionError", { configId: Schema.String }, @@ -37,6 +42,7 @@ export class ServiceFailureError extends Schema.TaggedError export type Error = | SessionNotFoundError + | SessionDirectoryMismatchError | InvalidConfigOptionError | InvalidModelError | InvalidEffortError @@ -49,6 +55,11 @@ export function toRequestError(error: Error): RequestError { switch (error._tag) { case "ACPSessionNotFoundError": return RequestError.invalidParams({ sessionId: error.sessionId }, `session not found: ${error.sessionId}`) + case "ACPSessionDirectoryMismatchError": + return RequestError.invalidParams( + { sessionId: error.sessionId, cwd: error.cwd }, + `session ${error.sessionId} does not belong to cwd: ${error.cwd}`, + ) case "ACPInvalidConfigOptionError": return RequestError.invalidParams({ configId: error.configId }, `unknown config option: ${error.configId}`) case "ACPInvalidModelError": diff --git a/packages/cli/src/acp/event.ts b/packages/cli/src/acp/event.ts index ef6bea22e27..5943e6cc84b 100644 --- a/packages/cli/src/acp/event.ts +++ b/packages/cli/src/acp/event.ts @@ -1,10 +1,11 @@ -import type { AgentSideConnection, PromptResponse, SessionUpdate } from "@agentclientprotocol/sdk" +import type { PromptResponse, SessionUpdate } from "@agentclientprotocol/sdk" import type { EventSubscribeOutput, OpenCodeClient, SessionMessageAssistant, SessionMessageInfo, } from "@opencode/client/promise" +import type { ACPConnection } from "./connection" import { partsToContentChunks, type ReplayPart } from "./content" import { ACPError } from "./error" import { replyPermission, syncEditedFiles } from "./permission" @@ -17,11 +18,11 @@ import { type ToolInput, } from "./tool" -type Connection = Pick & - Partial> +type Connection = Pick export type TurnControl = { cancelled: boolean + // Aborted whenever the turn is cancelled or closed, so it also cancels the turn's outbound client requests. readonly admission: AbortController } @@ -160,6 +161,7 @@ export async function streamTurn(input: { clientSessionID: input.sessionID, cwd: input.cwd, tool, + signal: control.admission.signal, ...(child ? { toolCallPrefix: child.id, titlePrefix: child.title } : {}), }) continue @@ -269,6 +271,7 @@ export async function streamTurn(input: { toolName: current.name, toolInput: current.input, metadata: event.data.metadata ?? {}, + signal: control.admission.signal, }).catch(() => {}) await send({ sessionUpdate: "tool_call_update", @@ -417,7 +420,7 @@ function projectChildUpdate(update: SessionUpdate, child: ChildSession) { } export async function replayMessages( - connection: Pick, + connection: Pick, sessionID: string, cwd: string, messages: readonly SessionMessageInfo[], @@ -426,7 +429,7 @@ export async function replayMessages( } async function replayMessage( - connection: Pick, + connection: Pick, sessionID: string, cwd: string, message: SessionMessageInfo, diff --git a/packages/cli/src/acp/permission.ts b/packages/cli/src/acp/permission.ts index 826afd867fa..ab62a877077 100644 --- a/packages/cli/src/acp/permission.ts +++ b/packages/cli/src/acp/permission.ts @@ -1,12 +1,13 @@ -import type { AgentSideConnection, PermissionOption, ToolCallContent, ToolCallLocation } from "@agentclientprotocol/sdk" +import type { PermissionOption, ToolCallContent, ToolCallLocation } from "@agentclientprotocol/sdk" import type { EventSubscribeOutput, OpenCodeClient } from "@opencode/client/promise" import { Patch } from "@opencode/util/patch" import { Result } from "effect" import { isAbsolute, resolve } from "node:path" +import type { ACPConnection } from "./connection" import { pendingToolCall, stringValue, toLocations, toToolKind, type ToolInput } from "./tool" type PermissionEvent = Extract -type Connection = Pick & Partial> +type Connection = Pick type Tool = { readonly name: string; readonly input: ToolInput } const options: PermissionOption[] = [ @@ -25,31 +26,34 @@ export async function replyPermission(input: { readonly tool?: Tool readonly toolCallPrefix?: string readonly titlePrefix?: string + readonly signal?: AbortSignal }) { const toolName = input.tool?.name ?? input.event.data.action const toolInput = { ...input.event.data.metadata, ...input.tool?.input } const previews = await permissionPreviews(toolName, toolInput, input.cwd) const toolCallID = input.event.data.source?.id ?? input.event.data.id const title = permissionTitle(toolName, toolInput, previews) - const result = await input.connection - .requestPermission({ - sessionId: input.clientSessionID ?? input.sessionID, - toolCall: { - ...pendingToolCall({ - toolCallId: input.toolCallPrefix ? `${input.toolCallPrefix}:${toolCallID}` : toolCallID, - toolName, - state: { - input: toolInput, - title: prefixedTitle(input.titlePrefix, title), - }, - cwd: input.cwd, - }), - locations: permissionLocations(toolName, toolInput, input.event.data.resources, input.cwd, previews), - ...(previews.length > 0 ? { content: previews } : {}), - }, - options, - }) - .catch(() => undefined) + const request = { + sessionId: input.clientSessionID ?? input.sessionID, + toolCall: { + ...pendingToolCall({ + toolCallId: input.toolCallPrefix ? `${input.toolCallPrefix}:${toolCallID}` : toolCallID, + toolName, + state: { + input: toolInput, + title: prefixedTitle(input.titlePrefix, title), + }, + cwd: input.cwd, + }), + locations: permissionLocations(toolName, toolInput, input.event.data.resources, input.cwd, previews), + ...(previews.length > 0 ? { content: previews } : {}), + }, + options, + } + // An already-cancelled turn skips the round-trip; the SDK would still send the request and then cancel it. + const result = input.signal?.aborted + ? undefined + : await input.connection.requestPermission(request, { cancellationSignal: input.signal }).catch(() => undefined) const selected = result?.outcome.outcome === "selected" ? result.outcome.optionId : undefined const reply = selected === "once" || selected === "always" ? selected : "reject" await input.client.permission.reply({ @@ -66,13 +70,14 @@ function prefixedTitle(prefix: string | undefined, title: string | undefined) { } export async function syncEditedFiles(input: { - readonly connection: Partial> + readonly connection: Pick readonly writeTextFile: boolean readonly sessionID: string readonly cwd: string readonly toolName: string readonly toolInput: ToolInput readonly metadata: Readonly> + readonly signal?: AbortSignal }) { if (!input.writeTextFile || !input.connection.writeTextFile || toToolKind(input.toolName) !== "edit") return const files = Array.isArray(input.metadata.files) @@ -89,7 +94,10 @@ export async function syncEditedFiles(input: { const target = resolvePath(path, input.cwd) const file = Bun.file(target) if (!(await file.exists())) return - await input.connection.writeTextFile?.({ sessionId: input.sessionID, path: target, content: await file.text() }) + await input.connection.writeTextFile?.( + { sessionId: input.sessionID, path: target, content: await file.text() }, + { cancellationSignal: input.signal }, + ) }), ) } diff --git a/packages/cli/src/acp/service.ts b/packages/cli/src/acp/service.ts index c835cbc1f3c..349d7b39e24 100644 --- a/packages/cli/src/acp/service.ts +++ b/packages/cli/src/acp/service.ts @@ -7,9 +7,9 @@ import { type SessionInfo, type SessionMessageInfo, } from "@opencode/client/promise" +import { FSUtil } from "@opencode/util/fs-util" import { withTimestampedFallback } from "@opencode/util/session-title-fallback" import type { - AgentSideConnection, AuthenticateRequest, AuthenticateResponse, AuthMethod, @@ -46,6 +46,7 @@ import { parseModelSelection, type ConfigOptionProvider, } from "./config-option" +import type { ACPConnection } from "./connection" import { promptContentToParts } from "./content" import { ChildSessionUpdateMethod, @@ -60,9 +61,6 @@ import { ACPError } from "./error" export const AuthMethodID = "opencode-login" -type Connection = Pick & - Partial> - type Catalog = { readonly providers: ConfigOptionProvider[] readonly models: ModelInfo[] @@ -102,15 +100,18 @@ export interface Interface { forkSession(input: ForkSessionRequest): Promise setSessionConfigOption(input: SetSessionConfigOptionRequest): Promise setSessionMode(input: SetSessionModeRequest): Promise - prompt(input: PromptRequest): Promise + prompt(input: PromptRequest, signal?: AbortSignal): Promise cancel(input: CancelNotification): Promise } -export function make(input: { readonly client: OpenCodeClient; readonly connection: Connection }): Interface { +export function make(input: { + readonly client: OpenCodeClient + readonly connection: ACPConnection.Connection +}): Interface { const sessions = new Map() const catalogs = new Map>() const registeredMcp = new Map>() - const active = new Map() + const active = new Map }>() const capabilities = { writeTextFile: false, childSessionUpdates: false } const catalog = (cwd: string) => { @@ -136,6 +137,15 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti registeredMcp.delete(sessionID) } + const cancelTurn = (sessionID: string) => { + const turn = active.get(sessionID) + if (turn) { + turn.control.cancelled = true + turn.control.admission.abort() + } + return input.client.session.interrupt({ sessionID }) + } + const attach = async (session: SessionInfo, cwd: string, mcpServers: readonly McpServer[]) => { const currentCatalog = await catalog(cwd) sessions.get(session.id)?.abort.abort() @@ -216,7 +226,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti return { sessionId: state.id, configOptions: configOptions(state) } }, loadSession: async (params) => { - const session = await getSession(input.client, params.sessionId) + const session = await getSession(input.client, params.sessionId, params.cwd) const state = await attach(session, session.location.directory, params.mcpServers) await replay(state) return { configOptions: configOptions(state) } @@ -246,18 +256,17 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti return {} }, resumeSession: async (params) => { - const session = await getSession(input.client, params.sessionId) + const session = await getSession(input.client, params.sessionId, params.cwd) const state = await attach(session, session.location.directory, params.mcpServers ?? []) return { configOptions: configOptions(state) } }, closeSession: async (params) => { - detach(params.sessionId) const turn = active.get(params.sessionId) - if (turn) { - turn.cancelled = true - turn.admission.abort() - } - await input.client.session.interrupt({ sessionID: params.sessionId }).catch(() => {}) + await cancelTurn(params.sessionId).catch((error) => { + if (!isSessionNotFoundError(error)) throw error + }) + await turn?.turn.catch(() => {}) + detach(params.sessionId) return {} }, forkSession: async (params) => { @@ -303,7 +312,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti await selectMode(input.client, await requireSession(params.sessionId), params.modeId) return {} }, - prompt: async (params) => { + prompt: async (params, signal) => { const state = await requireSession(params.sessionId) if (active.has(state.id)) { throw new ACPError.ServiceFailureError({ @@ -319,8 +328,9 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti capabilities.childSessionUpdates && extNotification ? (update: ChildSessionUpdate) => extNotification(ChildSessionUpdateMethod, update).then(() => {}) : undefined - active.set(state.id, control) - const response = await streamTurn({ + // A `$/cancel_request` for this prompt behaves like `session/cancel` for its turn. + const cancel = () => void cancelTurn(state.id).catch(() => {}) + const turn = streamTurn({ client: input.client, connection: input.connection, sessionID: state.id, @@ -333,19 +343,23 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti sessionSignal: state.abort.signal, submit: (signal) => submitPrompt(input.client, state, prepared, signal), ...(childSessionUpdate ? { childSessionUpdate } : {}), - }).finally(() => { - if (active.get(state.id) === control) active.delete(state.id) }) - await sendUsageUpdate(input.client, input.connection, state, response.usage?.totalTokens).catch(() => {}) - return response + .then(async (response) => { + await sendUsageUpdate(input.client, input.connection, state, response.usage?.totalTokens).catch(() => {}) + return response + }) + .finally(() => { + signal?.removeEventListener("abort", cancel) + if (active.get(state.id)?.control === control) active.delete(state.id) + }) + active.set(state.id, { control, turn }) + signal?.addEventListener("abort", cancel, { once: true }) + // The cancel may already be buffered behind the awaits above. + if (signal?.aborted) cancel() + return turn }, cancel: async (params) => { - const current = active.get(params.sessionId) - if (current) { - current.cancelled = true - current.admission.abort() - } - await input.client.session.interrupt({ sessionID: params.sessionId }).catch(() => {}) + await cancelTurn(params.sessionId).catch(() => {}) }, } } @@ -472,11 +486,15 @@ async function selectMode(client: OpenCodeClient, state: Attached, modeID: strin await client.session.switchAgent({ sessionID: state.id, agent: modeID }) } -async function getSession(client: OpenCodeClient, sessionID: string) { - return client.session.get({ sessionID }).catch((error) => { +async function getSession(client: OpenCodeClient, sessionID: string, cwd: string) { + const session = await client.session.get({ sessionID }).catch((error) => { if (isSessionNotFoundError(error)) throw new ACPError.SessionNotFoundError({ sessionId: sessionID }) throw error }) + if (FSUtil.resolve(cwd) !== FSUtil.resolve(session.location.directory)) { + throw new ACPError.SessionDirectoryMismatchError({ sessionId: sessionID, cwd }) + } + return session } async function messages(client: OpenCodeClient, sessionID: string) { @@ -542,7 +560,12 @@ function stableStringify(value: unknown): string { .join(",")}}` } -async function sendUsageUpdate(client: OpenCodeClient, connection: Connection, session: Attached, used?: number) { +async function sendUsageUpdate( + client: OpenCodeClient, + connection: ACPConnection.Connection, + session: Attached, + used?: number, +) { if (!used) return const model = session.catalog.models.find( (item) => item.providerID === session.model.providerID && item.id === session.model.id, diff --git a/packages/cli/src/commands/handlers/acp.ts b/packages/cli/src/commands/handlers/acp.ts index 9170b646eed..e170f05bc06 100644 --- a/packages/cli/src/commands/handlers/acp.ts +++ b/packages/cli/src/commands/handlers/acp.ts @@ -1,4 +1,4 @@ -import { AgentSideConnection, ndJsonStream } from "@agentclientprotocol/sdk" +import { ndJsonStream } from "@agentclientprotocol/sdk" import { OpenCode } from "@opencode/client/promise" import { Service } from "@opencode/client/effect/service" import { Effect } from "effect" @@ -26,8 +26,7 @@ export default Runtime.handler( process.stdin.on("error", (error) => controller.error(error)) }, }) - const stream = ndJsonStream(input, output) - const connection = new AgentSideConnection((connection) => ACP.create(client, connection), stream) + const connection = ACP.connect(client, ndJsonStream(input, output)) process.stdin.resume() yield* Effect.promise(() => connection.closed) // EOF owns this stdio process; exiting also closes the private server's lease pipe. diff --git a/packages/cli/test/acp/cancel.subprocess.test.ts b/packages/cli/test/acp/cancel.subprocess.test.ts new file mode 100644 index 00000000000..aa42e1285f3 --- /dev/null +++ b/packages/cli/test/acp/cancel.subprocess.test.ts @@ -0,0 +1,84 @@ +import type { CloseSessionResponse, PromptResponse } from "@agentclientprotocol/sdk" +import { describe, expect, test } from "bun:test" +import { createAcpFixture, expectOk, initialize, newSession } from "./subprocess" + +// The mock model holds any completion whose request mentions "hold" until the test releases it, +// so cancellation can be exercised while the model turn is genuinely in flight. +function heldModel() { + const started = Promise.withResolvers() + const release = Promise.withResolvers() + return { + started: started.promise, + release: () => release.resolve("released"), + respond(request: unknown) { + if (!JSON.stringify(request).includes("hold")) return "accepted" + started.resolve() + return release.promise + }, + } +} + +describe("acp cancel subprocess", () => { + test("$/cancel_request cancels the in-flight prompt and the session stays usable", async () => { + const model = heldModel() + await using fixture = await createAcpFixture({ respond: model.respond }) + const acp = fixture.spawn() + await initialize(acp) + const session = await newSession(acp, fixture.home) + + const prompt = acp.send("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "hold" }], + }) + await model.started + await acp.notify("$/cancel_request", { requestId: prompt.id }) + + expect(expectOk(await prompt.response).stopReason).toBe("cancelled") + model.release() + const next = expectOk( + await acp.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "again" }], + }), + ) + expect(next.stopReason).toBe("end_turn") + }, 60_000) + + test("session/close settles the active prompt before responding and leaves other sessions alone", async () => { + const model = heldModel() + await using fixture = await createAcpFixture({ respond: model.respond }) + const acp = fixture.spawn() + await initialize(acp) + const closing = await newSession(acp, fixture.home) + const other = await newSession(acp, fixture.home) + + const prompt = acp.send("session/prompt", { + sessionId: closing.sessionId, + prompt: [{ type: "text", text: "hold" }], + }) + await model.started + const order: string[] = [] + const close = acp + .request("session/close", { sessionId: closing.sessionId }) + .then((response) => { + order.push("close") + return response + }) + const cancelled = await prompt.response.then((response) => { + order.push("prompt") + return response + }) + + expect(expectOk(cancelled).stopReason).toBe("cancelled") + expect(expectOk(await close)).toEqual({}) + expect(order).toEqual(["prompt", "close"]) + model.release() + const next = expectOk( + await acp.request("session/prompt", { + sessionId: other.sessionId, + prompt: [{ type: "text", text: "still here" }], + }), + ) + expect(next.stopReason).toBe("end_turn") + }, 60_000) +}) diff --git a/packages/cli/test/acp/lifecycle.subprocess.test.ts b/packages/cli/test/acp/lifecycle.subprocess.test.ts index e8280aec4ae..1e03b80be85 100644 --- a/packages/cli/test/acp/lifecycle.subprocess.test.ts +++ b/packages/cli/test/acp/lifecycle.subprocess.test.ts @@ -51,6 +51,12 @@ describe("acp lifecycle subprocess", () => { ) expect(selectConfigOption(loaded.configOptions, "model")?.category).toBe("model") + const mismatched = await acp.request("session/load", { + cwd: fixture.root, + sessionId: session.sessionId, + mcpServers: [], + }) + expect(mismatched.error?.code).toBe(-32602) }, 60_000) test("list request includes a live ACP-created session", async () => { diff --git a/packages/cli/test/acp/permission-behavior.test.ts b/packages/cli/test/acp/permission-behavior.test.ts index d14c69007ca..ad8ab1e63eb 100644 --- a/packages/cli/test/acp/permission-behavior.test.ts +++ b/packages/cli/test/acp/permission-behavior.test.ts @@ -1,20 +1,25 @@ import { describe, expect, test } from "bun:test" -import type { AgentSideConnection, RequestPermissionRequest, RequestPermissionResponse } from "@agentclientprotocol/sdk" +import type { + RequestPermissionRequest, + RequestPermissionResponse, + SessionNotification, + WriteTextFileRequest, +} from "@agentclientprotocol/sdk" import fs from "node:fs/promises" import os from "node:os" import path from "node:path" +import type { ACPConnection } from "../../src/acp/connection" import { streamTurn } from "../../src/acp/event" import { syncEditedFiles } from "../../src/acp/permission" import { createSseFixture, durableEvent, ephemeralEvent, withTimeout } from "./sse-fixture" -type SessionUpdateParams = Parameters[0] -type Connection = Pick & - Partial> +type SessionUpdateParams = SessionNotification +type Connection = Pick type Fixture = ReturnType describe("acp permission behavior", () => { test("does not sync edits when writeTextFile was not advertised", async () => { - const writes: Parameters[0][] = [] + const writes: WriteTextFileRequest[] = [] await syncEditedFiles({ connection: { @@ -216,7 +221,7 @@ describe("acp permission behavior", () => { const file = path.join(cwd, "file.ts") await fs.writeFile(file, "before") const permissionRequests: RequestPermissionRequest[] = [] - const writes: Parameters[0][] = [] + const writes: WriteTextFileRequest[] = [] const fixture = createSseFixture({ onPrompt({ id, send }) { send(durableEvent("session.inbox.delivered", { sessionID: "ses_edit", inboxID: id })) @@ -306,7 +311,7 @@ describe("acp permission behavior", () => { "*** End Patch", ].join("\n") const permissionRequests: RequestPermissionRequest[] = [] - const writes: Parameters[0][] = [] + const writes: WriteTextFileRequest[] = [] const fixture = createSseFixture({ onPrompt({ id, send }) { send(durableEvent("session.inbox.delivered", { sessionID: "ses_patch", inboxID: id })) @@ -536,6 +541,55 @@ describe("acp permission behavior", () => { await fixture.stop() } }) + + test("cancelling the turn cancels its pending permission request and rejects the permission", async () => { + const requested = Promise.withResolvers() + const fixture = createSseFixture({ + onPrompt({ id, send }) { + send(durableEvent("session.inbox.delivered", { sessionID: "ses_cancel", inboxID: id })) + send(permissionAsked("ses_cancel", "perm_cancel")) + }, + onPermissionReply({ send }) { + send(durableEvent("session.execution.interrupted", { sessionID: "ses_cancel", reason: "user" })) + }, + }) + const connection = { + sessionUpdate: async () => {}, + // Behaves like a client answering the agent's `$/cancel_request` with a cancelled outcome. + requestPermission: (_request, options) => + new Promise((resolve) => { + options?.cancellationSignal?.addEventListener("abort", () => resolve({ outcome: { outcome: "cancelled" } }), { + once: true, + }) + requested.resolve(options?.cancellationSignal) + }), + } satisfies Connection + const control = { cancelled: false, admission: new AbortController() } + const result = streamTurn({ + client: fixture.client, + connection, + sessionID: "ses_cancel", + cwd: "/workspace", + start: { type: "input", id: "input_cancel" }, + writeTextFile: false, + control, + submit: (signal) => + fixture.client.session.prompt({ sessionID: "ses_cancel", id: "input_cancel", text: "hello" }, { signal }), + }) + + try { + const signal = await withTimeout(requested.promise, "permission was not requested") + expect(signal?.aborted).toBe(false) + control.cancelled = true + control.admission.abort() + + expect(await withTimeout(result, "cancelled turn did not finish")).toMatchObject({ stopReason: "cancelled" }) + expect(permissionReplies(fixture)).toEqual([["perm_cancel", "reject"]]) + } finally { + await result.catch(() => undefined) + await fixture.stop() + } + }) }) function startTurn(fixture: Fixture, connection: Connection, sessionID: string, inboxID: string, cwd = "/workspace") { diff --git a/packages/cli/test/acp/service-fixture.ts b/packages/cli/test/acp/service-fixture.ts index 2e3e1909b35..d7f4866a06d 100644 --- a/packages/cli/test/acp/service-fixture.ts +++ b/packages/cli/test/acp/service-fixture.ts @@ -1,4 +1,4 @@ -import type { AgentSideConnection } from "@agentclientprotocol/sdk" +import type { SessionNotification } from "@agentclientprotocol/sdk" import { OpenCode, type AgentInfo, @@ -114,7 +114,7 @@ export function makeSession( export function makeACPFixture(options: FixtureOptions = {}) { const requests: FixtureRequest[] = [] - const updates: Parameters[0][] = [] + const updates: SessionNotification[] = [] const encoder = new TextEncoder() let eventController: ReadableStreamDefaultController | undefined const models = options.models ?? [testModel, secondModel] diff --git a/packages/cli/test/acp/service-lifecycle.test.ts b/packages/cli/test/acp/service-lifecycle.test.ts index 833631a04b3..6b87904977d 100644 --- a/packages/cli/test/acp/service-lifecycle.test.ts +++ b/packages/cli/test/acp/service-lifecycle.test.ts @@ -94,20 +94,24 @@ describe("acp service lifecycle", () => { }) const loaded = await fixture.service.loadSession({ - cwd: "/ignored", + cwd: "/workspace", sessionId: "ses_loaded", mcpServers: [], }) const resumed = await fixture.service.resumeSession({ - cwd: "/ignored", + cwd: "/workspace", sessionId: "ses_resume", mcpServers: [], }) const forked = await fixture.service.forkSession({ - cwd: "/ignored", + cwd: "/workspace", sessionId: "ses_loaded", mcpServers: [], }) + const mismatched = await fixture.service + .loadSession({ cwd: "/elsewhere", sessionId: "ses_loaded", mcpServers: [] }) + .catch((error: unknown) => error) + expect(mismatched).toMatchObject({ _tag: "ACPSessionDirectoryMismatchError", cwd: "/elsewhere" }) expect(currentValue(loaded, "model")).toBe("test/second-model") expect(currentValue(loaded, "effort")).toBe("medium") @@ -214,7 +218,8 @@ describe("acp service lifecycle", () => { ]) }) - test("cancel preserves the attachment while close removes it and interrupts best-effort", async () => { + test("cancel preserves the attachment while close removes it and surfaces interrupt failures", async () => { + const interrupt = { fail: true } await using fixture = makeACPFixture({ fetch(request) { if (request.method === "POST" && request.path === "/api/session") { @@ -223,8 +228,14 @@ describe("acp service lifecycle", () => { if (request.method === "POST" && request.path === "/api/session/ses_lifecycle/model") { return new Response(null, { status: 204 }) } + if (request.method === "POST" && request.path === "/api/session/missing/interrupt") { + return Response.json( + { _tag: "SessionNotFoundError", sessionID: "missing", message: "session not found" }, + { status: 404 }, + ) + } if (request.method === "POST" && request.path.endsWith("/interrupt")) { - return new Response(null, { status: 500 }) + return interrupt.fail ? new Response(null, { status: 500 }) : Response.json({ interrupted: false }) } return undefined }, @@ -239,6 +250,9 @@ describe("acp service lifecycle", () => { }) expect(currentValue(updated, "effort")).toBe("high") + const failed = await fixture.service.closeSession({ sessionId: created.sessionId }).catch((error: unknown) => error) + expect(failed).toMatchObject({ name: "ClientError", reason: "UnexpectedStatus" }) + interrupt.fail = false expect(await fixture.service.closeSession({ sessionId: created.sessionId })).toEqual({}) const missing = await fixture.service .setSessionConfigOption({ @@ -252,6 +266,7 @@ describe("acp service lifecycle", () => { expect( fixture.requests.filter((request) => request.path.endsWith("/interrupt")).map((request) => request.path), ).toEqual([ + "/api/session/ses_lifecycle/interrupt", "/api/session/ses_lifecycle/interrupt", "/api/session/ses_lifecycle/interrupt", "/api/session/missing/interrupt", diff --git a/packages/cli/test/acp/subprocess.ts b/packages/cli/test/acp/subprocess.ts index 584a986e4b6..1032c541767 100644 --- a/packages/cli/test/acp/subprocess.ts +++ b/packages/cli/test/acp/subprocess.ts @@ -46,6 +46,11 @@ type Waiter = { export type AcpProcess = { readonly request: (method: string, params?: unknown) => Promise> + readonly send: ( + method: string, + params?: unknown, + ) => { readonly id: number; readonly response: Promise> } + readonly notify: (method: string, params: unknown) => Promise readonly waitForNotification: ( method: string, predicate: (params: T) => boolean, @@ -64,7 +69,9 @@ description: Verifier compatibility skill. # Verifier Skill ` -export async function createAcpFixture(options: { readonly skill?: string } = {}) { +export async function createAcpFixture( + options: { readonly skill?: string; readonly respond?: (request: unknown) => string | Promise } = {}, +) { const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-cli-acp-")) const home = path.join(root, "workspace") const config = path.join(root, "config") @@ -84,8 +91,9 @@ export async function createAcpFixture(options: { readonly skill?: string } = {} if (request.method !== "POST" || new URL(request.url).pathname !== "/v1/chat/completions") { return new Response("Not found", { status: 404 }) } - requests.push(await request.json().catch(() => undefined)) - return new Response(completion("accepted"), { + const body: unknown = await request.json().catch(() => undefined) + requests.push(body) + return new Response(completion(await (options.respond?.(body) ?? "accepted")), { headers: { "content-type": "text/event-stream" }, }) }, @@ -294,18 +302,28 @@ function spawnAcp(input: { readonly env: Record }): }) } - return { - async request(method: string, params?: unknown) { - if (inputClosed) throw new Error("ACP stdin is closed") - const id = nextID++ - const request: JsonRpcRequest = - params === undefined ? { jsonrpc: "2.0", id, method } : { jsonrpc: "2.0", id, method, params } - await child.stdin.write(encoder.encode(`${JSON.stringify(request)}\n`)) - await child.stdin.flush() + const write = async (message: JsonRpcRequest | JsonRpcNotification) => { + if (inputClosed) throw new Error("ACP stdin is closed") + await child.stdin.write(encoder.encode(`${JSON.stringify(message)}\n`)) + await child.stdin.flush() + } + + const send = (method: string, params?: unknown) => { + const id = nextID++ + const request: JsonRpcRequest = + params === undefined ? { jsonrpc: "2.0", id, method } : { jsonrpc: "2.0", id, method, params } + const response = write(request).then(async () => { const response = await take((message) => isResponse(message) && message.id === id, 20_000, `${method} response`) if (!isResponse(response)) throw new Error(`Invalid ACP response: ${JSON.stringify(response)}`) return response - }, + }) + return { id, response } + } + + return { + request: (method: string, params?: unknown) => send(method, params).response, + send, + notify: (method: string, params: unknown) => write({ jsonrpc: "2.0", method, params }), async waitForNotification(method: string, predicate: (params: T) => boolean, timeoutMs = 20_000) { const notification = await take( (message) => isNotification(message) && message.method === method && predicate(message.params),